Skip to content

Pipelines & Worker — the pipeline axis

The sample proves the full loop: API creates a job document → worker polls → pipeline family runs → job reaches a terminal status.

The jobs slice

domain/jobs/ is an ordinary CRUD slice; its model is the durable state machine the worker drives:

1
2
3
4
5
6
7
class Job(BaseModel):
    name: str
    kind: str                                        # dispatch key
    status: Literal["pending", "processing", "done", "failed"] = "pending"
    payload: dict | None = None
    retry_count: int = 0
    error_message: str | None = None

POST /jobs with {"name": "...", "kind": "appointment_reminder", "payload": {"appointment_id": 1}} enqueues work; GET /jobs?status=... inspects the queue.

The appointment_reminder family

1
2
3
4
5
6
7
8
9
pipelines/appointment_reminder/
    abc.py            # LoadPort / ComposePort / DeliverPort + root interface
    impl.py           # concrete stages, wired root, register("appointment_reminder", ...)
    pipeline.yaml     # graph spec (nodes, edges, roots)
    resources/
        pipeline.yaml # runtime spec consumed by the loader
        patterns/reminder.yaml   # message template fragments
pipelines/shared/
    ports.py loader.py registry.py   # reused by every family

Each stage is a small, independently testable unit. The loader (pipelines/shared/loader.py) wires the root through hexa.build and merges resources/patterns/*.yaml into the stage-injected registry:

1
2
pipeline = get_appointment_reminder_pipeline(appointment_svc)
reminder = await pipeline.run(appointment_id)

To add a family: copy appointment_reminder/, replace the stages, call register(kind, variant, root_cls) — see Pipeline Scaffold.

The worker

worker.py is the process that turns documents into execution:

  1. poll job_svc.search({"status": "pending"}) on WORKER_POLL_SECONDS
  2. claim a job: status → processing
  3. dispatch through the HANDLERS table — HANDLERS[job.kind](job, resources)
  4. on success status → done; on exception status → failed, retry_count += 1, error_message captured; jobs at WORKER_MAX_RETRIES are skipped
1
2
3
4
HANDLERS: dict[str, Callable] = {
    "appointment_reminder": run_appointment_reminder,
    # add yours here
}

The worker shares the composition root: from init import init gives it the same wired services the API uses — one place to add a dependency.

Same process model in production

API and worker are separate processes (see docker-compose.yaml: api, worker, mongo). Any in-memory progress bus is per-process; to surface live progress through the API, swap the bus adaptor for a broker-backed one — see Worker & Progress.

Tests

  • tests/unit/test_appointment_reminder_pipeline.py — stage units + the wired root against a fake service.
  • tests/unit/test_worker.py — dispatch, done/failed transitions, retry exhaustion, using an in-memory job service double and a cancelled poll task.