From 91e3c6824c0c3686b3b8038e1e802deccb8ee62c Mon Sep 17 00:00:00 2001 From: Donald Gray Date: Fri, 31 Jul 2026 16:13:58 +0100 Subject: [PATCH 1/3] Replace unmaintained django-q with django-q2 django-q 1.3.9 has been unmaintained since 2021; django-q2 is the maintained drop-in fork (same django_q import path and INSTALLED_APPS entry). Ships additional django_q migrations, applied by the existing 'manage.py migrate' step. Prerequisite for routing tasks to multiple queues via ALT_CLUSTERS. Co-Authored-By: Claude Fable 5 --- src/requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/requirements.txt b/src/requirements.txt index 3fd699a..690e8b1 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -4,7 +4,7 @@ boto3==1.34.144 Django==4.2.14 django-environ==0.11.2 django-health-check==3.18.3 -django-q==1.3.9 +django-q2==1.10.0 djangorestframework==3.15.2 gunicorn==22.0.0 jsonschema==4.23.0 From 208b7977d89f30045f02b2229e1580ac649018ff Mon Sep 17 00:00:00 2001 From: Donald Gray Date: Fri, 31 Jul 2026 16:14:09 +0100 Subject: [PATCH 2/3] Add priority queue for expedited submissions POST /customers/{customer}/queue/priority accepts the same payload as the standard endpoint but routes processing through a VIP lane at both levels, mirroring the DLCS API's own priority queue: - locally, tasks are enqueued to a dedicated queue/cluster (named by PRIORITY_QUEUE_NAME) with its own worker pool, so they are not blocked by a standard-queue backlog. Each engine container runs a second qcluster process for it; no new instances are required. - at ingest, batches for priority collections are POSTed to the DLCS /queue/priority endpoint. If PRIORITY_QUEUE_NAME is unset the feature is inert: priority submissions use the standard local queue but still ingest via the DLCS priority endpoint. See docs/priority-queue.md for the full design. Also repairs the integration test stack, which referenced a Dockerfile.CompositeHandler left over from the protagonist monorepo import and was unrunnable: builds the root Dockerfile instead, supplies the required env vars, gates services on a postgres healthcheck, and bumps pytest-docker for Docker Compose v2. Co-Authored-By: Claude Fable 5 --- .env.dist | 5 + README.md | 25 +- docker-compose.local.yml | 2 - docker-compose.yml | 2 - docs/priority-queue.md | 254 ++++++++++++++++++ entrypoints/entrypoint-worker.sh | 11 +- src/app/api/urls.py | 4 + src/app/api/views.py | 15 +- src/app/common/dlcs.py | 5 +- .../migrations/0005_collection_priority.py | 18 ++ src/app/common/models.py | 1 + src/app/engine/tasks.py | 7 +- src/app/settings.py | 14 + src/tests/api/test_api.py | 19 ++ src/tests/docker-compose.yml | 46 ++-- src/tests/pytest_requirements.txt | 6 +- 16 files changed, 403 insertions(+), 31 deletions(-) create mode 100644 docs/priority-queue.md create mode 100644 src/app/common/migrations/0005_collection_priority.py diff --git a/.env.dist b/.env.dist index 323f21f..a42a7ba 100644 --- a/.env.dist +++ b/.env.dist @@ -37,6 +37,11 @@ ENGINE_WORKER_MAX_ATTEMPTS=0 # Django Q SQS Broker SQS_BROKER_QUEUE_NAME=composite-handler-queue +# Priority queue (blank = disabled; with the SQS broker this must be the name +# of a real SQS queue) +PRIORITY_QUEUE_NAME= +PRIORITY_WORKER_COUNT=1 + # Run migrations MIGRATE=True diff --git a/README.md b/README.md index 874284c..9975350 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ The DLCS Composite Handler is an implementation of [DLCS RFC011](https://github. The component is written in Python and utilises Django with the following extensions: - [Django REST Framework](https://github.com/encode/django-rest-framework/tree/master) -- [Django Q](https://github.com/Koed00/django-q) +- [Django Q2](https://github.com/django-q2/django-q2) - [django-environ](https://github.com/joke2k/django-environ) - [django-health-check](https://github.com/KristianOellegaard/django-health-check) @@ -92,18 +92,35 @@ The following list of environment variables are supported: | `MIGRATE` | None | API, Engine | If "True" will run migrations + createcachetable on startup if entrypoint used. | | `INIT_SUPERUSER` | None | API, Engine | If "True" will attempt to create superuser. Needs standard Django envvars to be set (e.g. `DJANGO_SUPERUSER_USERNAME`, `DJANGO_SUPERUSER_EMAIL`, `DJANGO_SUPERUSER_PASSWORD`) if entrypoint used. | | `GUNICORN_WORKERS` | `2` | API | The value of [`--workers`](https://docs.gunicorn.org/en/stable/run.html) arg when running gunicorn | -| `SQS_BROKER_QUEUE_NAME` | None | API, Engine | If set, django-q [SQS broker](https://django-q.readthedocs.io/en/latest/brokers.html#amazon-sqs) will be used. Queue created if doesn't exist. If empty default [Django ORM broker](https://django-q.readthedocs.io/en/latest/brokers.html#django-orm) is used | +| `SQS_BROKER_QUEUE_NAME` | None | API, Engine | If set, django-q [SQS broker](https://django-q2.readthedocs.io/en/master/brokers.html#amazon-sqs) will be used. Queue created if doesn't exist. If empty default [Django ORM broker](https://django-q2.readthedocs.io/en/master/brokers.html#django-orm) is used | +| `PRIORITY_QUEUE_NAME` | None | API, Engine | If set, enables the priority queue: submissions to `/customers/{customer}/queue/priority` are routed to a dedicated queue/cluster of this name. When the SQS broker is in use, this must be the name of a real SQS queue. If empty, priority submissions use the standard queue. | +| `PRIORITY_WORKER_COUNT` | `1` | Engine | The number of workers spawned by each engine instance for the priority cluster. | Note that in order to access the S3 bucket, the Composite Handler assumes that valid AWS credentials are available in the environment - this can be in the former of [environment variables](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html), or in the form of ambient credentials. ### Django Q Broker -By default Django Q will use the default [Django ORM](https://django-q.readthedocs.io/en/latest/brokers.html#django-orm) broker. +By default Django Q will use the default [Django ORM](https://django-q2.readthedocs.io/en/master/brokers.html#django-orm) broker. -The [SQS broker](https://django-q.readthedocs.io/en/latest/brokers.html#amazon-sqs) can be configured by specifying the `SQS_BROKER_QUEUE_NAME` environment variable. Default SQS broker behaviour is to create this queue if it is not found. +The [SQS broker](https://django-q2.readthedocs.io/en/master/brokers.html#amazon-sqs) can be configured by specifying the `SQS_BROKER_QUEUE_NAME` environment variable. Default SQS broker behaviour is to create this queue if it is not found. As with S3, above, Composite Handler assumes that valid AWS credentials are available in the environment. +### Priority Queue + +Submissions can be expedited by POSTing to `/customers/{customer}/queue/priority` instead of `/customers/{customer}/queue` — same body, same auth, same response. This mirrors the DLCS API's own priority queue. A priority submission: + +- is enqueued to a dedicated local queue (named by `PRIORITY_QUEUE_NAME`) with its own worker pool, so it is not blocked by a backlog on the standard queue, and +- is ingested into DLCS via `/customers/{customer}/queue/priority`, so it also skips any DLCS-side backlog. + +Each engine instance runs a second `qcluster` process for the priority queue (see [`entrypoint-worker.sh`](entrypoints/entrypoint-worker.sh)); to run one manually: + +```bash +Q_CLUSTER_NAME= python manage.py qcluster +``` + +If `PRIORITY_QUEUE_NAME` is unset, the feature degrades gracefully: priority submissions are accepted and processed via the standard local queue, but are still ingested through the DLCS priority endpoint. See [`priority-queue.md`](priority-queue.md) for the full design. + ## Building The project ships with a [`Dockerfile`](./Dockerfile): diff --git a/docker-compose.local.yml b/docker-compose.local.yml index 1b7dfbc..b4a39ff 100644 --- a/docker-compose.local.yml +++ b/docker-compose.local.yml @@ -1,5 +1,3 @@ -version: "3.9" - services: postgres: image: postgres:13.3 diff --git a/docker-compose.yml b/docker-compose.yml index 247f757..9cdaa01 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,5 +1,3 @@ -version: "3.9" - services: init: image: busybox:latest diff --git a/docs/priority-queue.md b/docs/priority-queue.md new file mode 100644 index 0000000..abef77d --- /dev/null +++ b/docs/priority-queue.md @@ -0,0 +1,254 @@ +# Priority Queue — Design + +> Written by Claude (Claude Code), 2026-07-31, alongside the implementation it describes. +> +> Supersedes `expedite.md` and `expedite_simple.md`. Both contained useful halves of the +> answer but neither works alone; this document corrects the factual errors in the former +> and closes the gap in the latter. Verified against the pinned dependencies, the +> django-q2 source (v1.10.0), and the DLCS protagonist API. + +## Goal + +A VIP lane for PDF submissions. Processing is identical to the standard queue; the only +difference is that a priority submission must not wait behind a backlog (e.g. 100 queued +items) in either the composite-handler **or** DLCS itself. Anyone with valid credentials +may use the lane — no per-customer gating. Additional compute cost must be ~zero. + +## Why the previous documents don't work as written + +**`expedite.md` (Option A, `ALT_CLUSTERS`)** — two factual errors: + +1. This repo pins `django-q==1.3.9` (original django-q, unmaintained since 2021). + `ALT_CLUSTERS` does not exist in that library — it is a feature of **django-q2**, the + maintained drop-in fork. A library migration is a prerequisite the document never + mentions. +2. Even in django-q2, a single `qcluster` process does **not** monitor multiple queues + with a combined worker pool. One `qcluster` process serves exactly one cluster; an + alternate cluster is a *separate process*, selected at startup via the + `Q_CLUSTER_NAME` environment variable (`django_q/conf.py`; docs "Multiple Queues"). + `ALT_CLUSTERS` is only a dict of per-cluster config overrides keyed by that name. + +So Option A quietly collapses into the "two processes" shape of Option B. That is still +cheap (see [Cost](#cost)) — but the design below is written around how the library +actually behaves. + +**`expedite_simple.md` (DLCS `/queue/priority` only)** — solves half the problem. It +routes the *DLCS ingest step* to DLCS's priority queue but changes nothing about the +composite-handler's own queue. In the motivating scenario (100 items queued locally), a +priority submission still waits behind 100 fetch/rasterize/upload jobs before the DLCS +call ever happens. However, its plumbing (URL shape, `priority` flag on `Collection`, +pass-through to `dlcs.ingest`) is needed regardless, because DLCS has its own queue that +can also back up. + +## Design + +Two complementary halves, one submission flag: + +1. **Local VIP lane** — a second queue with a small dedicated worker pool, so priority + tasks skip the composite-handler backlog. Implemented with django-q2 `ALT_CLUSTERS` + plus a second `qcluster` process inside the **existing** engine containers. +2. **DLCS VIP lane** — priority submissions are ingested via + `POST /customers/{customer}/queue/priority` (verified present in protagonist's + `CustomerQueueController`: *"The processing is the same but the priority queue is for + images that need to be processed quickly"*), so they also skip any DLCS-side backlog. + +### API surface + +``` +POST /customers/{customer}/queue → standard (unchanged) +POST /customers/{customer}/queue/priority → VIP lane +``` + +Same request body, same auth, same response shape. Mirroring DLCS's own URL scheme keeps +the composite-handler a faithful proxy of DLCS semantics (RFC011) and makes priority use +auditable in access logs. No header or schema change needed. + +### Routing mechanics (verified against django-q2 v1.10.0 source) + +- `async_task(..., cluster=)` → `get_broker()` → for the SQS broker, + `` **is the SQS queue name** (`brokers/__init__.py`: `list_key` == + queue name; `tasks.py`: `get_broker(task.get("cluster"))`). +- A worker process started with `Q_CLUSTER_NAME=` merges + `ALT_CLUSTERS[]` over the root `Q_CLUSTER` config and consumes from queue + `` (`conf.py`: `CLUSTER_NAME = conf.get("cluster_name", PREFIX)`). +- With the ORM broker (local dev without SQS), the same `cluster=` value keys rows + in the `OrmQ` table and routing works identically — no SQS required for dev/tests. +- `cluster=None` routes to the default cluster, i.e. today's behaviour. + +### Graceful degradation + +If `PRIORITY_QUEUE_NAME` is not configured, priority submissions are accepted, enqueue to +the **standard** local queue, and still ingest via DLCS `/queue/priority`. This makes the +rollout safe (API can deploy before engine/infra) and keeps the DLCS half working even if +the local lane is ever disabled. + +## Changes + +### 0. Prerequisite: migrate `django-q` → `django-q2` (standalone step) + +- `src/requirements.txt`: replace `django-q==1.3.9` with `django-q2==1.10.0` + (same `django_q` import path and `INSTALLED_APPS` entry — no code changes; + pulls in `django-picklefield`; supports Django ≥4.2, Python 3.9–3.13). +- `python manage.py migrate` — django-q2 ships additional `django_q` migrations + (already run by `entrypoint.sh` when `MIGRATE=True`). +- **Ship and deploy this alone first.** It swaps an unmaintained 2021 library running + against Django 4.2 for the maintained fork — worth doing even without this feature — + and isolates the riskiest change so any fallout is unambiguous. + +### 1. Settings — `src/app/settings.py` + +```python +PRIORITY_QUEUE_NAME = env.str("PRIORITY_QUEUE_NAME", default="") + +if PRIORITY_QUEUE_NAME: + Q_CLUSTER["ALT_CLUSTERS"] = { + PRIORITY_QUEUE_NAME: { + "workers": env("PRIORITY_WORKER_COUNT", cast=int, default=1), + }, + } +``` + +The alt cluster inherits everything else (timeout, retry, max_attempts, broker choice) +from the root `Q_CLUSTER`. With SQS in use, `PRIORITY_QUEUE_NAME` must equal the name of +a real SQS queue (to be created in infra, mirroring the standard one). + +### 2. Model — `src/app/common/models.py` (+ migration) + +```python +class Collection(models.Model): + ... + priority = models.BooleanField(default=False) +``` + +`CollectionSerializer` needs **no change** — it uses `fields = "__all__"`. + +### 3. URLs — `src/app/api/urls.py` + +```python +path("customers//queue", CollectionAPIView.as_view()), +path("customers//queue/priority", CollectionAPIView.as_view(priority=True)), +``` + +### 4. View — `src/app/api/views.py` + +```python +class CollectionAPIView(AbstractAPIView): + priority = False + + def post(self, request, *args, **kwargs): + ... + serializer = CollectionSerializer( + data={ + "json_data": request.data, + "customer": kwargs["customer"], + "priority": self.priority, + } + ) + ... + for serializer in serializers: + member = serializer.save() + async_task( + "app.engine.tasks.process_member", + {"id": member.id, "auth": request.headers["Authorization"]}, + task_name=f"Submission: [{member.id}]", + cluster=settings.PRIORITY_QUEUE_NAME or None if self.priority else None, + ) +``` + +Optionally include `"priority": collection.priority` in +`_build_collection_response_body` for observability. + +### 5. DLCS client — `src/app/common/dlcs.py` + +```python +def ingest(self, customer, json, auth, priority=False): + endpoint = "queue/priority" if priority else "queue" + response = requests.post( + f"{self._api_root}customers/{customer}/{endpoint}", + ... + ) +``` + +### 6. Engine task — `src/app/engine/tasks.py` + +`__initiate_dlcs_ingest` passes `priority=member.collection.priority` to `dlcs.ingest`. + +### 7. Worker entrypoint — `entrypoints/entrypoint-worker.sh` + +Run both clusters in the existing engine container; exit (→ container restart) if either +dies. Behaviour is unchanged when `PRIORITY_QUEUE_NAME` is unset: + +```bash +bash entrypoint.sh + +if [[ -n "$PRIORITY_QUEUE_NAME" ]]; then + python manage.py qcluster & + Q_CLUSTER_NAME="$PRIORITY_QUEUE_NAME" python manage.py qcluster & + wait -n + exit 1 +else + python manage.py qcluster +fi +``` + +### 8. Config templates & docs + +- `.env.dist`: add `PRIORITY_QUEUE_NAME=` (blank = disabled) and + `PRIORITY_WORKER_COUNT=1` under the Django Q section. +- `README.md`: document the endpoint, the two env vars, and the SQS queue requirement. + +### Infrastructure (outside this repo) + +- One new SQS queue, named to match `PRIORITY_QUEUE_NAME`, same settings as the standard + queue (visibility timeout must remain ≥ `ENGINE_WORKER_RETRY`). +- Set `PRIORITY_QUEUE_NAME` / `PRIORITY_WORKER_COUNT` on the engine and API services. +- **No new instances, services, or replicas.** + +## Cost + +The marginal footprint is one extra `qcluster` process group per engine instance +(sentinel + 1 worker + monitor + pusher — a few hundred MB RAM, ~zero CPU while idle). +Idle priority workers do not lend capacity to the standard queue; that is the point — +they are always free when a VIP job lands. When a VIP job runs it shares the instance's +CPU with standard rasterization work, which is acceptable: the problem being solved is +*queue wait* (hours behind a backlog), not execution speed. + +If even that idle footprint must go, the same code supports a scale-from-zero deployment +(a separate engine service running only the priority cluster, autoscaled 0→1 on SQS queue +depth). That is purely a deployment change — not designed here, and not recommended as a +starting point. + +## Testing + +Integration tests (`src/tests/`, pytest-docker, ORM broker): + +- `POST /customers/{c}/queue/priority` returns 202, persists `Collection.priority=True`, + and reports `"priority": true` in the response; the standard endpoint reports `false`. +- Existing suite must pass unchanged after the django-q2 swap (step 0). +- The `dlcs.ingest` endpoint switch has no automated coverage (the test stack runs no + engine); it is verified by inspection/mock — see implementation notes below. + +Note: the test stack's `docker-compose.yml` referenced a `Dockerfile.CompositeHandler` +that never existed in this repo (a leftover from the protagonist monorepo import), so the +integration tests were unrunnable as imported. Fixed as part of this work: the compose +file now builds the root `Dockerfile` and supplies the required env vars, and +`pytest-docker` is bumped to a version that uses Docker Compose v2. + +## Rollout order + +1. **PR 1**: django-q2 migration only. Deploy, soak. +2. **Infra**: create the priority SQS queue. +3. **PR 2**: everything else. Deployable before or after the env vars are set — the + feature is inert until `PRIORITY_QUEUE_NAME` is configured, and priority submissions + degrade gracefully (standard local queue + DLCS priority ingest) in the meantime. + +## Decisions & open items + +- **Access control**: none — any authenticated caller may use the lane (matches DLCS's + own priority queue). If abuse ever makes the VIP queue back up, per-customer gating can + be added at the view layer later. +- **Priority worker count**: default 1 per engine instance (3 with the current 3 + replicas). Tune via `PRIORITY_WORKER_COUNT` only if VIP volume grows. +- **In-flight work is not preempted**: a VIP job skips the queue but still waits for a + free priority worker; with `workers: 1` per instance, concurrent VIP jobs queue behind + each other. Acceptable for the stated "rare, needed by CoP" use case. diff --git a/entrypoints/entrypoint-worker.sh b/entrypoints/entrypoint-worker.sh index 615e3b1..1a7932c 100644 --- a/entrypoints/entrypoint-worker.sh +++ b/entrypoints/entrypoint-worker.sh @@ -4,4 +4,13 @@ set -o errexit set -o pipefail bash entrypoint.sh -python manage.py qcluster + +if [[ -n "$PRIORITY_QUEUE_NAME" ]]; then + python manage.py qcluster & + Q_CLUSTER_NAME="$PRIORITY_QUEUE_NAME" python manage.py qcluster & + # If either cluster dies, exit so the container is restarted with both. + wait -n + exit 1 +else + python manage.py qcluster +fi diff --git a/src/app/api/urls.py b/src/app/api/urls.py index 51c12fe..296aa10 100644 --- a/src/app/api/urls.py +++ b/src/app/api/urls.py @@ -8,4 +8,8 @@ QueryMemberAPIView.as_view(), ), path("customers//queue", CollectionAPIView.as_view()), + path( + "customers//queue/priority", + CollectionAPIView.as_view(priority=True), + ), ] diff --git a/src/app/api/views.py b/src/app/api/views.py index b51d9ea..6274ed1 100644 --- a/src/app/api/views.py +++ b/src/app/api/views.py @@ -34,6 +34,7 @@ def _validate_credentials(self, customer, headers): def _build_collection_response_body(self, collection): return { "id": f"{self._scheme}://{self._hostname}/collections/{collection.id}", + "priority": collection.priority, "members": [ self._build_member_response_body(member) for member in Member.objects.filter(collection=collection) @@ -97,11 +98,17 @@ def get(self, request, *args, **kwargs): class CollectionAPIView(AbstractAPIView): + priority = False + def post(self, request, *args, **kwargs): self._validate_credentials(kwargs["customer"], request.headers) serializer = CollectionSerializer( - data={"json_data": request.data, "customer": kwargs["customer"]} + data={ + "json_data": request.data, + "customer": kwargs["customer"], + "priority": self.priority, + } ) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @@ -115,12 +122,18 @@ def post(self, request, *args, **kwargs): if not all(serializer.is_valid() for serializer in serializers): return Response(status=status.HTTP_400_BAD_REQUEST) + cluster = ( + settings.PRIORITY_QUEUE_NAME + if self.priority and settings.PRIORITY_QUEUE_NAME + else None + ) for serializer in serializers: member = serializer.save() async_task( "app.engine.tasks.process_member", {"id": member.id, "auth": request.headers["Authorization"]}, task_name=f"Submission: [{member.id}]", + cluster=cluster, ) return Response( diff --git a/src/app/common/dlcs.py b/src/app/common/dlcs.py index 56482f5..1917364 100644 --- a/src/app/common/dlcs.py +++ b/src/app/common/dlcs.py @@ -23,9 +23,10 @@ def test_credentials(self, customer, auth): else: response.raise_for_status() - def ingest(self, customer, json, auth): + def ingest(self, customer, json, auth, priority=False): + endpoint = "queue/priority" if priority else "queue" response = requests.post( - f"{self._api_root}customers/{customer}/queue", + f"{self._api_root}customers/{customer}/{endpoint}", headers={"Content-Type": "application/json", "Authorization": auth}, json=json, ) diff --git a/src/app/common/migrations/0005_collection_priority.py b/src/app/common/migrations/0005_collection_priority.py new file mode 100644 index 0000000..e874541 --- /dev/null +++ b/src/app/common/migrations/0005_collection_priority.py @@ -0,0 +1,18 @@ +# Generated by Django 4.2.14 on 2026-07-31 14:44 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('common', '0004_auto_20220113_1314'), + ] + + operations = [ + migrations.AddField( + model_name='collection', + name='priority', + field=models.BooleanField(default=False), + ), + ] diff --git a/src/app/common/models.py b/src/app/common/models.py index 36b6754..8ab9262 100644 --- a/src/app/common/models.py +++ b/src/app/common/models.py @@ -7,6 +7,7 @@ class Collection(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) json_data = models.JSONField() customer = models.IntegerField() + priority = models.BooleanField(default=False) created_date = models.DateTimeField(auto_now_add=True) last_updated_date = models.DateTimeField(auto_now=True) diff --git a/src/app/engine/tasks.py b/src/app/engine/tasks.py index 3ae7901..4829300 100644 --- a/src/app/engine/tasks.py +++ b/src/app/engine/tasks.py @@ -69,7 +69,12 @@ def __initiate_dlcs_ingest(member, dlcs_requests, auth): dlcs_responses = [] for dlcs_request in dlcs_requests: dlcs_responses.append( - dlcs.ingest(member.collection.customer, dlcs_request, auth) + dlcs.ingest( + member.collection.customer, + dlcs_request, + auth, + priority=member.collection.priority, + ) ) return dlcs_responses diff --git a/src/app/settings.py b/src/app/settings.py index 1ae1cb8..f7d0280 100644 --- a/src/app/settings.py +++ b/src/app/settings.py @@ -164,6 +164,20 @@ else: Q_CLUSTER["orm"] = "default" +# When set, priority submissions are routed to a dedicated cluster of this +# name (which, when the SQS broker is in use, must match the name of a real +# SQS queue). A worker for it is started with `Q_CLUSTER_NAME= python +# manage.py qcluster`. When unset, priority submissions fall back to the +# standard queue. +PRIORITY_QUEUE_NAME = env.str("PRIORITY_QUEUE_NAME", default="") + +if PRIORITY_QUEUE_NAME: + Q_CLUSTER["ALT_CLUSTERS"] = { + PRIORITY_QUEUE_NAME: { + "workers": env("PRIORITY_WORKER_COUNT", cast=int, default=1), + }, + } + SCRATCH_DIRECTORY = env.path("SCRATCH_DIRECTORY", default="/tmp/scratch") WEB_SERVER = { diff --git a/src/tests/api/test_api.py b/src/tests/api/test_api.py index 3828539..806323e 100644 --- a/src/tests/api/test_api.py +++ b/src/tests/api/test_api.py @@ -110,6 +110,7 @@ def test_collection_api_view(http_service): act_vals = r.json() assert act_vals["id"] + assert act_vals["priority"] is False act_members = act_vals["members"] assert len(act_members) == 1 @@ -120,6 +121,24 @@ def test_collection_api_view(http_service): assert "last_updated" in act_member +def test_collection_api_view_priority(http_service): + r = requests.post( + f"{http_service}/customers/123/queue/priority", + headers=test_headers, + json=test_collection, + ) + assert r.status_code == 202 + + act_vals = r.json() + assert act_vals["id"] + assert act_vals["priority"] is True + + act_members = act_vals["members"] + assert len(act_members) == 1 + for act_member in act_members: + assert act_member["status"] == "PENDING" + + def test_collection_query(http_service): collection_id = test_fixtures["collection"]["id"] diff --git a/src/tests/docker-compose.yml b/src/tests/docker-compose.yml index 5afd2cb..e72aeac 100644 --- a/src/tests/docker-compose.yml +++ b/src/tests/docker-compose.yml @@ -1,49 +1,65 @@ -version: "3.9" - services: migrate: build: - dockerfile: ../Dockerfile.CompositeHandler - context: ../ + dockerfile: Dockerfile + context: ../.. command: bash -c "python manage.py migrate && python manage.py createcachetable" environment: + - DJANGO_SECRET_KEY=integration-test-secret-key - DATABASE_URL=postgresql://dlcs:password@postgres:5432/compositedb + - CACHE_URL=dbcache://app_cache depends_on: - - postgres + postgres: + condition: service_healthy load_test_data: build: - dockerfile: ../Dockerfile.CompositeHandler - context: ../ + dockerfile: Dockerfile + context: ../.. command: bash -c "python manage.py loaddata /srv/dlcs/test_fixtures/dump.json" environment: + - DJANGO_SECRET_KEY=integration-test-secret-key - DATABASE_URL=postgresql://dlcs:password@postgres:5432/compositedb + - CACHE_URL=dbcache://app_cache volumes: - type: bind source: ./api/test_fixtures target: /srv/dlcs/test_fixtures depends_on: - - postgres - - migrate + postgres: + condition: service_healthy + migrate: + condition: service_completed_successfully api: build: - dockerfile: ../Dockerfile.CompositeHandler - context: ../ + dockerfile: Dockerfile + context: ../.. command: python manage.py runserver 0.0.0.0:8000 environment: + - DJANGO_SECRET_KEY=integration-test-secret-key - DATABASE_URL=postgresql://dlcs:password@postgres:5432/compositedb + - CACHE_URL=dbcache://app_cache - DLCS_API_ROOT=http://dlcs:8080/ ports: - "8000" depends_on: - - migrate - - postgres - - dlcs - - load_test_data + postgres: + condition: service_healthy + migrate: + condition: service_completed_successfully + load_test_data: + condition: service_completed_successfully + dlcs: + condition: service_started postgres: image: postgres:13.3 environment: - POSTGRES_USER=dlcs - POSTGRES_PASSWORD=password - POSTGRES_DB=compositedb + healthcheck: + test: ["CMD-SHELL", "pg_isready -U dlcs -d compositedb"] + interval: 2s + timeout: 3s + retries: 30 dlcs: image: mendhak/http-https-echo:22 diff --git a/src/tests/pytest_requirements.txt b/src/tests/pytest_requirements.txt index ddd05f2..121fe2a 100644 --- a/src/tests/pytest_requirements.txt +++ b/src/tests/pytest_requirements.txt @@ -1,3 +1,3 @@ -pytest==6.2.5 -pytest-docker==0.10.3 -requests==2.26.0 +pytest==9.1.1 +pytest-docker==3.2.5 +requests==2.32.3 From 5631c191f0a9b71fec7753924a40da0f97fbbc36 Mon Sep 17 00:00:00 2001 From: Donald Gray Date: Fri, 31 Jul 2026 16:27:17 +0100 Subject: [PATCH 3/3] Black format --- src/app/common/migrations/0005_collection_priority.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/app/common/migrations/0005_collection_priority.py b/src/app/common/migrations/0005_collection_priority.py index e874541..522c3b2 100644 --- a/src/app/common/migrations/0005_collection_priority.py +++ b/src/app/common/migrations/0005_collection_priority.py @@ -4,15 +4,14 @@ class Migration(migrations.Migration): - dependencies = [ - ('common', '0004_auto_20220113_1314'), + ("common", "0004_auto_20220113_1314"), ] operations = [ migrations.AddField( - model_name='collection', - name='priority', + model_name="collection", + name="priority", field=models.BooleanField(default=False), ), ]