Jobs API

Job retrieval

queue.get_job()

queue.get_job(job_id: str) -> JobResult | None

Retrieve a job by ID. Returns None if not found.

queue.list_jobs()

queue.list_jobs(
    status: str | None = None,
    queue: str | None = None,
    task_name: str | None = None,
    limit: int = 50,
    offset: int = 0,
    namespace: str | None = _UNSET,
) -> list[JobResult]

List jobs with optional filters. Returns newest first. Defaults to the queue's namespace — pass namespace=None to see all namespaces.

queue.list_jobs_filtered()

queue.list_jobs_filtered(
    status: str | None = None,
    queue: str | None = None,
    task_name: str | None = None,
    metadata_like: str | None = None,
    error_like: str | None = None,
    created_after: float | None = None,
    created_before: float | None = None,
    limit: int = 50,
    offset: int = 0,
    namespace: str | None = _UNSET,
) -> list[JobResult]

Extended filtering with metadata and error pattern matching and time range constraints. Defaults to the queue's namespace — pass namespace=None to see all namespaces.

queue.list_jobs_after() / queue.list_jobs_filtered_after()

queue.list_jobs_after(..., limit: int = 50, after: str | None = None) -> Page[JobResult]
queue.list_jobs_filtered_after(..., limit: int = 50, after: str | None = None) -> Page[JobResult]

The keyset-paginated twins of the two listings above: same filters, offset swapped for after. Each returns a Page — iterate it directly for the rows, and pass its next_cursor back as after for the following page. A next_cursor of None means that was the last page. Treat the cursor as opaque; do not build or parse one.

Use these to walk a listing rather than page it. An offset re-scans every row it skips, so a deep page costs more than a shallow one, and a job enqueued mid-walk shifts every later offset by one — rows get seen twice or missed entirely. A keyset page is O(page) at any depth and stable under concurrent inserts.

Job operations

queue.cancel_job()

queue.cancel_job(job_id: str) -> bool

Cancel a pending job. Returns True if cancelled, False if not pending. Cascade-cancels dependents.

queue.cancel_running_job()

queue.cancel_running_job(job_id: str) -> bool

Set the cancel flag on a job that has already started, since cancel_job() only works before dispatch. True means the request was recorded, not that anything stopped: cancellation is cooperative. The handler observes it by calling current_job.check_cancelled(), which raises TaskCancelledError; a task that never checks runs to completion.

Declared as request_cancel, and check_cancelled() reads the flag back through is_cancel_requested. Under the prefork pool the child installs a local cancel check first, so an out-of-band cancel is observed without a storage round-trip.

queue.update_progress()

queue.update_progress(job_id: str, progress: int) -> None

Update progress for a running job (0–100).

queue.job_errors()

queue.job_errors(job_id: str) -> list[dict]

Every failed attempt for a job, oldest first — a list of dicts with id, job_id, attempt, error, failed_at. Retries append rather than replace, so this is the record of why a job reached the dead-letter queue, not just what the last attempt said. Empty for a job that has never failed. Declared as get_job_errors.

queue.job_dag()

queue.job_dag(job_id: str) -> dict

Return a dependency graph for a job, including all ancestors and descendants. Useful for visualizing workflow chains.

Archival

Archiving moves terminal jobs out of the hot jobs table into archived_jobs, so listings and dispatch scans stay fast without losing the history. The archive has its own retention window — see Retention.

queue.archive()

queue.archive(older_than: int = 86400) -> int

Archive every completed, dead or cancelled job older than older_than seconds, returning how many moved. Age-based and bulk: there is no per-job form. Declared as archive_old_jobs.

queue.list_archived() / queue.list_archived_after()

queue.list_archived(limit: int = 50, offset: int = 0) -> list[JobResult]
queue.list_archived_after(limit: int = 50, after: str | None = None) -> Page[JobResult]

Read the archive back, newest first. list_archived() is offset-paginated and fine for one page; list_archived_after() is the keyset twin — pass the previous page's next_cursor back as after, and a next_cursor of None means that was the last page.

Prefer the keyset form for a full walk. An offset re-scans everything it skips, and a job archived mid-walk shifts every later offset by one, so rows get seen twice or not at all. Cursors are opaque — do not build or parse one. Async twins: alist_archived() and alist_archived_after().

Replay

queue.replay()

queue.replay(job_id: str) -> JobResult

Re-enqueue a job with its original arguments, and record the replay. A new job, not a resumed one: it gets a fresh id and a fresh retry budget, and the original row is left exactly as it was. Declared as replay_job, which returns that new id — the Python method fetches it back as a JobResult for you.

There is no status check, so a pending or running job replays as readily as a failed one — replaying a job that is still in flight gives you two of it. The copy also drops the original's unique_key, its dependencies and its expiry, so a replay is not deduplicated against the job it came from and does not wait on anything the original waited on.

queue.replay_history()

queue.replay_history(job_id: str) -> list[dict]

Every replay of a job, newest first — each dict carries id, original_job_id, replay_job_id, replayed_at (Unix ms), original_error and replay_error. It is the audit trail for "we retried this by hand", which is otherwise invisible: a replay creates an unrelated-looking job. Declared as get_replay_history.