queueflow

Python client for QueueFlow, a PostgreSQL-native distributed job queue and workflow engine.

Typed end-to-end: request/response models (pydantic v2) mirror the server's OpenAPI 3.1 spec.
Ergonomic: qf.create_job(...), wait_for_* pollers, and a workflow builder DSL.
Worker runtime: qf.run_worker(...) runs Python task handlers against a remote server.
Thin facade over a generated core: the facade (queueflow.facade) adds what codegen cannot; the generated queueflow.api.* clients and queueflow.models.* stay available for everything else.

Python 3.9 or newer.

Install

pip install queueflow

Quick start

from queueflow.facade import QueueFlow, wf

qf = QueueFlow("http://localhost:8000", "dev")

# Enqueue a job and wait for the result.
job = qf.create_job("echo", payload={"hello": "world"}, max_retries=3, timeout=30)
done = qf.wait_for_job(job.id)
print(done.status, done.result)   # completed {'hello': 'world', 'echoed': True}

# Declare and run a DAG workflow.
dag = (
    wf("etl")
    .step("extract", "echo")
    .step("transform", "echo", after=["extract"])
    .step("load", "echo", after=["transform"], on_failure="halt")
)
workflow = qf.create_workflow(dag)
finished = qf.wait_for_workflow(workflow.id)
print(finished.status, finished.context)

echo is a handler built into the server; it returns the payload plus "echoed": true.

API

Client

qf = QueueFlow(
    base_url,             # required, e.g. "http://localhost:8000"
    token,                # required: tenant bearer token (any non-empty token on the dev server)
    worker_token=None,    # credential for qf.worker and run_worker (defaults to token; dev mode only)
    timeout=30.0,         # default per-request timeout in seconds; a (connect, read) pair or None also work
)

Any generated api method accepts _request_timeout= to override the default for one call.

The generated per-tag clients hang off the facade for anything the helpers do not cover: qf.jobs, qf.workflows, qf.worker, qf.cron, qf.dlq, qf.system, qf.health_api.

Jobs

Method	Description
`qf.create_job(task, payload=None, priority=None, max_retries=None, timeout=None, queue=None, retry_backoff=None, retry_delay_secs=None, retry_max_delay_secs=None, jitter_factor=None, idempotency_key=None, run_at=None)`	Enqueue a job and return the created `Job`.
`qf.wait_for_job(job_id, timeout=60.0, interval=0.5)`	Poll until `completed` / `failed` / `cancelled`; raises `WaitTimeout`.
`qf.jobs.get_job(id)`	Fetch a job.
`qf.jobs.list_jobs(status=..., queue=..., task_name=..., limit=..., offset=..., order_by=..., include_total=..., cursor=..., created_after=..., created_before=...)`	List jobs.
`qf.jobs.cancel_job(id)`	Cancel a job.
`qf.cancel_jobs(queue=None, task_name=None, status=None, created_after=None, created_before=None, limit=None)`	Cancel every matching non-terminal job of your tenant, oldest first, up to `limit` (default 100); returns the count.
`qf.jobs.create_batch_jobs(CreateBatchJobsRequest(...))`	Enqueue up to 1000 jobs at once.

idempotency_key is sent as the Idempotency-Key header: re-submitting the same key returns the original job instead of creating a duplicate. run_at (a datetime) delays the first run.

List responses carry next_cursor when there are more pages; pass it back as cursor for keyset pagination (cheaper than deep offset).

cancel_jobs takes optional filters that combine: queue, task_name, status (pending, retrying or running; all three when omitted, a terminal status is a 400) and the half-open datetime window [created_after, created_before). A count below limit means the matching set is exhausted, so loop until then to drain a large backlog:

while qf.cancel_jobs(queue="orders", task_name="send_email", limit=500) == 500:
    pass

Workflows

Method	Description
`qf.create_workflow(builder_or_request)`	Create a workflow from a `wf()` builder or a raw `CreateWorkflowRequest`.
`qf.wait_for_workflow(workflow_id, timeout=60.0, interval=0.5)`	Poll until a terminal workflow state.
`qf.workflows.get_workflow(id)` / `list_workflows(...)` / `cancel_workflow(id)`	Fetch / list / cancel.
`qf.workflow_steps(workflow_id)`	Live progress of every step, in declaration order: `name`, `status`, `job_id` (the first child, for a fan-out), `error` (`condition not met` for a step skipped by its `when`), `started_at`, `completed_at`, `child_workflow_id` (sub-workflow steps) and `fan_out` (`total`, `completed`, `failed`).
`qf.retry_workflow(workflow_id)`	Re-run a `failed`, `partially_failed` or `cancelled` workflow from where it stopped and return its refreshed record (server 0.6.0). A running or completed workflow is a 409.
`qf.workflows.get_workflow_step_states(id)`	The raw step-states call behind `qf.workflow_steps`.
`qf.workflows.get_workflow_diagram(id)`	Mermaid (`graph TD`) diagram of the DAG.

The Workflow record carries status, the accumulated context (step results keyed by step name), the step definitions and, from server 0.6.0, error_message, timeout_secs and deadline_at, retry_count, and parent_workflow_id / parent_step_name on a child spawned by a sub-workflow step.

retry_workflow sends every step that did not complete (failed, cancelled, or skipped by a failure policy) back to pending; completed steps and condition-skipped steps keep their results and do not run again. The workflow becomes running, its error_message clears, retry_count increments and the deadline, if any, is re-armed from the original timeout_secs. Fix the cause first: a retry with the same inputs fails the same way.

Workflow builder

from queueflow.facade import wf

fulfilment = (
    wf("fulfilment")
    .step("pick", "pick_items", queue="warehouse")
    .step("pack", "pack_parcel", after=["pick"], queue="warehouse")
    .timeout_secs(1800)
)

dag = (
    wf("order_123")
    .step("validate", "validate_order", payload={"order_id": 123})
    .step("charge", "charge_payment", after=["validate"])
    .step("fraud", "fraud_check", after=["validate"], on_failure="halt")
    .step("review", "queue_for_review", after=["fraud"],
          when={"path": "fraud.flags", "op": "eq", "value": {"manual": True}})
    .step("resize", "resize_image", after=["validate"], queue="images", on_failure="continue",
          fan_out={"path": "validate.images", "item_key": "image"})
    .sub_workflow("fulfil", fulfilment, after=["charge", "fraud"], payload={"warehouse": "eu-1"})
    .step("confirm", "send_confirmation", after=["fulfil"], on_failure="skip")
    .context(source="web")
    .metadata(app="shop")
    .timeout_secs(3600)
)
# .build() runs locally first: duplicate names, dangling deps, cycles, and malformed
# when / fan_out shapes raise WorkflowValidationError before any network round-trip.

Builder method	Description
`step(name, task, after=None, payload=None, on_failure=None, on_success=None, config=None, queue=None, when=None, fan_out=None)`	A task step. `config` is a `models.JobConfig` override; `on_failure` is `halt` (default), `skip` or `continue`.
`sub_workflow(name, child, after=None, payload=None, on_failure=None, on_success=None, when=None)`	A step that runs `child` (a `wf()` builder, built and validated here, or a `CreateWorkflowRequest`) as an inline child workflow instead of a task.
`context(kwargs)` / `metadata(kwargs)`	Seed the shared context / store bookkeeping JSON on the workflow. Both accumulate across calls.
`timeout_secs(secs)`	Whole-workflow deadline. When it passes, every unfinished step is cancelled and the workflow ends `failed` with `error_message` `workflow timed out after <secs>s`.
`build()`	Validate locally and return the `CreateWorkflowRequest`; `qf.create_workflow` calls it for you.

queue, when, fan_out, sub_workflow and timeout_secs need server 0.6.0 or later.

Queues. queue= puts the step's job(s) on that queue (the server's default queue when omitted), so a step can be routed to the workers that know its task.

Conditions. when= gates a step on the workflow context at the moment it becomes ready. It takes a dict in the wire shape or a models.StepCondition. A leaf names a dotted path (step results live under their step name, so validate.ok is the ok key of validate's result) and compares the value there with value using op: eq (default with a value), ne, gt, gte, lt, lte, exists (default without a value), contains, in. {"all": [...]}, {"any": [...]} and {"not": {...}} combine leaves. A false condition marks the step skipped with the error condition not met; unlike a step skipped by a failure policy, its dependents still run, simply without a result for it in _context. With this release a leaf's value must be a JSON object (see Known limitations); exists needs no value.

Fan-out. fan_out= ({"path": ..., "item_key": "item", "max_items": 10000}, or a models.FanOut) runs the task once per element of the array at path, all children created in one atomic claim and run concurrently across the queue's workers, each with its own retries. A child's payload is the step payload plus the element under item_key and its position under _index; its metadata carries fan_out_index and fan_out_total. The step completes with the result {"results": [r0, r1, ...]} in element order (_context.<step>.results downstream); a failed child follows the step's on_failure. The step state's fan_out reports total, completed and failed, and its job_id is the first child.

Sub-workflows. sub_workflow() runs the child as a workflow of its own, in the same tenant, with parent_workflow_id and parent_step_name set; the parent's step state carries its child_workflow_id. The child's context is its declared context, then the step's payload, then the parent's whole context under _parent. A completed child completes the step with the child's final context as the step result; a failed, partially failed or cancelled child fails the step under its on_failure. Cancelling or timing out the parent cancels a running child. Children nest up to 8 levels and carry their own step queues.

Worker

Run task handlers in this process against a remote QueueFlow server:

import threading
from queueflow.facade import QueueFlow, NonRetryableError

qf = QueueFlow("http://localhost:8000", "tenant-key", worker_token="worker-secret")

def send_email(job):
    if not job.payload.get("to"):
        raise NonRetryableError("no recipient")   # straight to the dead-letter queue
    ...                                           # raising anything else retries per policy
    return {"sent": True}

def transcode(job, ctx):
    for chunk in chunks(job.payload):
        if ctx.cancelled:                         # lease lost / cancelled mid-run: stop,
            return {}                             # the server owns the outcome now
        process(chunk)
    return {"ok": True}

stop = threading.Event()                          # set it (e.g. from a signal handler) to drain
result = qf.run_worker(
    "orders",
    {"send_email": send_email, "transcode": transcode},
    stop=stop,
    concurrency=4,
)
for job in result.unfinished:                     # handlers still running when the drain gave up
    print("abandoned", job.id)

run_worker(
    queue, handlers,
    lease_secs=30,        # lease per job; each in-flight job is heartbeated at half this
    wait_secs=20,         # long-poll wait when the queue is empty
    stop=None,            # threading.Event: set it to stop
    on_error=None,        # called on transient lease errors (default: throttled log)
    concurrency=1,        # handlers running at once, on a ThreadPoolExecutor
    poll_wait_secs=5,     # cap on wait_secs while `stop` is provided (bounds stop latency)
    drain_timeout=30.0,   # seconds to wait for in-flight handlers after stop
) -> WorkerRunResult      # .unfinished: jobs still running when the drain gave up

run_worker heartbeats each in-flight job at half the lease interval, stops reporting when the lease is lost, and applies the server's retry policy on errors. Handlers take (job) or (job, ctx) and return the result dict (or None for {}). Delivery is at-least-once, so make handlers idempotent. The low-level calls (qf.worker.lease_jobs, heartbeat_job, complete_job, fail_job) are also exposed.

Concurrency: up to concurrency handlers run at once. Each lease asks for as many jobs as there are free slots, and no lease is issued while every slot is busy.

Stop and drain: set stop. The generated urllib3 client cannot cancel an in-flight long poll, so while stop is provided each lease waits at most poll_wait_secs (never more than wait_secs), which bounds stop latency to about that long; without stop, the full wait_secs is used. After the loop exits, handlers already running get up to drain_timeout seconds to finish and report. Any handler still running after that is abandoned: its ctx.cancelled flips, nothing is reported (the lease expires and the server redelivers), and the job is listed in the returned WorkerRunResult.unfinished. Python cannot kill a thread, so an abandoned handler keeps its pool thread until it returns; long handlers should check ctx.cancelled.

Timeouts: every request uses the client-wide timeout (30s by default, see Client), except the lease long poll, which gets its own timeout of the effective wait plus 35 seconds.

Rules the runtime enforces, and that any hand-rolled loop over qf.worker must honour too:

Worker routes authenticate with the worker token, not a tenant token. Pass it as worker_token=; without it qf.worker reuses the tenant token, which only works in the server's development mode. run_worker raises on 401/403 rather than spinning.
Every in-flight job is heartbeated at half its lease interval. A heartbeat showing the job no longer running (or an HTTP 409) abandons reporting and flips ctx.cancelled.
Raise NonRetryableError (or any exception with retryable = False) for permanent failures so they dead-letter immediately. A failed outcome report is never converted into a job failure; the lease expires and the engine redelivers.

Cron

Method	Description
`qf.create_cron(name, schedule, task, payload=None, queue=None)`	Register a recurring enqueue (5-field crontab, UTC); returns the schedule id.
`qf.cron.get_cron(id)` / `list_crons(...)` / `delete_cron(id)`	Fetch / list / delete.
`qf.cron.pause_cron(id)` / `resume_cron(id)`	Stop firings / resume at the next future occurrence.

Dead letters

Method	Description
`qf.dlq.list_dead_letters(task_name=..., replayed=..., limit=..., offset=..., order_by=..., include_total=..., cursor=..., created_after=..., created_before=...)` / `get_dead_letter(id)`	Inspect terminally-failed jobs. `replayed=False` lists only entries not yet replayed, `True` only the replayed ones.
`qf.replay_dead_letter(id)`	Re-run one as a fresh job and return it (at most once; a second replay is a 409).
`qf.replay_dead_letters(queue=None, task_name=None, created_after=None, created_before=None, limit=None)`	Replay every unreplayed entry matching the filter, oldest first, up to `limit` (default 100). Returns a `ReplayedDeadLetter` per entry: `job_id` on success, otherwise `error`.

for r in qf.replay_dead_letters(queue="orders", task_name="send_email"):
    if r.error:
        print("dead letter", r.dead_letter_id, "not replayed:", r.error)
    else:
        print("dead letter", r.dead_letter_id, "replayed as", r.job_id)

Already-replayed entries are never considered again, so repeating the call walks a large backlog; an empty list means nothing was left to replay.

System

qf.whoami()              # the tenant this token is scoped to and its access level
qf.system.get_stats()    # engine counters
qf.system.list_tasks()   # registered task handler names
qf.queues()              # live per-queue backlog: pending, scheduled, running, oldest pending age, paused
qf.pause_queue(name)     # stop claims of your jobs on a queue; running jobs finish
qf.resume_queue(name)    # let claims flow again
qf.health_api.get_health()
qf.health_api.get_ready()

qf.whoami() (GET /api/v1/whoami) returns a WhoAmIResponse with tenant and access (read_write, or read_only for a read-only key), handy for validating a credential at startup.

qf.queues() (GET /api/v1/queues) returns a QueueStats per queue so you can decide whether to add workers. A 0.5.0 or later server lists every queue your tenant has enqueued to (drained ones with zero counts); older servers list only queues with non-terminal jobs. Its paused flag says whether claims are paused for your tenant. qf.system.list_queues() is the raw call.

qf.pause_queue(name) / qf.resume_queue(name) (POST /api/v1/queues/{queue}/pause and /resume) gate workers' claims of your tenant's jobs on that queue. Pausing lets running jobs finish while pending ones wait, so the queue is drained once the returned running count reaches zero; resuming wakes parked workers immediately. Both are idempotent and return a QueueControlResponse (queue, paused, running, pending).

import time

qf.pause_queue("orders")
while qf.pause_queue("orders").running > 0:   # idempotent, so it doubles as a status poll
    time.sleep(1)
deploy_new_handlers()
qf.resume_queue("orders")

Authentication

Every request carries Authorization: Bearer <token>. Two credentials exist:

The tenant token (token) authenticates the job, workflow, cron, DLQ and system routes. On a server started without --api-keys, any non-empty token is accepted (development mode).
The worker token (worker_token) authenticates the worker-protocol routes (lease, heartbeat, complete, fail). Configure it on the server with --worker-token / QUEUEFLOW_WORKER_TOKEN. It is a separate secret, never a tenant token.

Error handling

Generated calls raise queueflow.exceptions.ApiException (with .status, .reason, .body) and its subclasses BadRequestException (400), UnauthorizedException (401), ForbiddenException (403), NotFoundException (404), ServiceException (5xx). The facade adds WaitTimeout (a wait_for_* deadline passed), WorkflowValidationError (a bad DAG, raised before any request), and NonRetryableError (raise from a handler to dead-letter immediately).

from queueflow.exceptions import ApiException, NotFoundException
from queueflow.facade import WaitTimeout

try:
    job = qf.wait_for_job("missing", timeout=10)
except NotFoundException:
    ...
except WaitTimeout:
    ...
except ApiException as err:
    print(err.status, err.body)

Known limitations

SSE streams buffer. JobsApi.stream_job_events and WorkflowsApi.stream_workflow_events cannot stream: the generated client buffers the whole SSE response until the server closes it (terminal status or the 15-minute cap). Use wait_for_job / wait_for_workflow (polling) instead, or the *_without_preload_content variants with a hand-rolled SSE parser.

Conformance tests

test/test_live.py runs the facade against a real QueueFlow server (started with the default queueflow serve, i.e. --mode all, so the built-in echo handler is registered). It creates and waits on an echo job, checks idempotent re-creation, runs a two-step workflow, exercises the cron and dead-letter endpoints, reads stats, runs run_worker against a dedicated queue, and exercises whoami, bulk cancel, queue pause/resume and bulk dead-letter replay, each scoped to a queue unique to the run. Against a 0.6.0 server it also runs its own worker on a run-unique queue to cover a condition-skipped step whose dependents still run, a fan-out with ordered results and progress counts, a sub-workflow child linked to its parent, and a retry after a non-retryable failure. The tests are marked live and skip themselves unless QUEUEFLOW_URL is set, so plain pytest and CI stay offline.

QUEUEFLOW_URL=http://localhost:8000 \
QUEUEFLOW_TOKEN=dev \
QUEUEFLOW_WORKER_TOKEN=worker-secret \
pytest -m live

QUEUEFLOW_TOKEN is the tenant token (default dev). QUEUEFLOW_WORKER_TOKEN is the server's --worker-token; it defaults to the tenant token, which only works when the server runs without one.

Architecture

This package is a thin hand-written facade over a generated core:

queueflow-sdk-python/
├── queueflow/
│   ├── facade.py        the hand-written facade (QueueFlow, wf, run_worker, errors)
│   ├── api/             generated per-tag clients (JobsApi, WorkflowsApi, WorkerApi, ...)
│   ├── models/          generated pydantic models
│   └── api_client.py, configuration.py, rest.py, exceptions.py   generated transport
├── docs/                generated per-endpoint and per-model reference
└── test/                test_facade.py (offline), test_live.py (conformance), generated model tests

The core is generated by openapi-generator from the server's OpenAPI spec (scripts/generate-sdks.sh python in queueflow-core). The facade is injected at generation time from sdk-templates/python/facade.mustache in that repo, so edit the template, not queueflow/facade.py. Per-endpoint reference for the generated layer lives in docs/.

Development

pip install -e . -r test-requirements.txt
pytest                 # offline tests (facade + generated model tests)
mypy                   # type-check queueflow/ and test/

Links

Documentation: docs.queueflow.dev
Server: github.com/elision-labs/queueflow-core
This SDK: github.com/elision-labs/queueflow-sdk-python

License

MIT

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
.openapi-generator		.openapi-generator
docs		docs
queueflow		queueflow
test		test
.gitignore		.gitignore
.openapi-generator-ignore		.openapi-generator-ignore
FACADE.md		FACADE.md
LICENSE		LICENSE
README.md		README.md
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt
setup.cfg		setup.cfg
setup.py		setup.py
test-requirements.txt		test-requirements.txt
tox.ini		tox.ini

Navigation Menu

Repository navigation

About

Repository files navigation

queueflow

Install

Quick start

API

Client

Jobs

Workflows

Workflow builder

Worker

Cron

Dead letters

System

Authentication

Error handling

Known limitations

Conformance tests

Architecture

Development

Links

License

About

Releases

Packages

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

queueflow

Install

Quick start

API

Client

Jobs

Workflows

Workflow builder

Worker

Cron

Dead letters

System

Authentication

Error handling

Known limitations

Conformance tests

Architecture

Development

Links

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages