Jobs & Clients¶

JobsClient is the primary entry point for enqueuing, querying, and cancelling jobs. It wraps a Backend and adds typed payload serialisation, JobHandle[R] construction, and CancelResult building.

Actor management: tq.actors provides list(), get(), set_capacity(), and deregister() — see Actor deregistration.

Job lifecycle¶

pending   → running   → succeeded
                       → failed        (retried → pending | terminal fail)
                       → cancelled
                       → crashed
                       → abandoned
                       → scheduled     (Snooze: job reschedules itself)

scheduled → pending   (scheduled_to_pending sweep, every ~1 s)

running   → scheduled (Snooze or RetryAfter with future scheduled_at)

Status	Meaning
`pending`	Waiting in the queue; eligible for dispatch.
`scheduled`	Enqueued with a future `scheduled_at`; not yet eligible.
`running`	Claimed by a worker; actor is executing.
`succeeded`	Actor returned successfully; result stored.
`failed`	Actor raised an unhandled exception and retry budget is exhausted, or `DeadlineExceeded`.
`cancelled`	Cancelled before or during execution.
`crashed`	Worker process died mid-execution (SIGKILL, OOM, etc.).
`abandoned`	Heartbeat expired and no worker reclaimed the job within the lock lease window.

Terminal statuses (succeeded, failed, cancelled, crashed, abandoned) have no further transitions.

Archival lifecycle. After a terminal job's per-status retention period elapses (default: 30–90 days depending on status), the maintenance leader's prune sweep moves it from jobs to jobs_archive. After the archive retention period elapses (default: 1 year), the archive expiry sweep hard-deletes the row. The admin UI job-detail page follows this chain automatically. See Configuration for retention settings.

Deferred jobs. Pass a timezone-aware scheduled_at datetime to delay execution. The job is stored with status="scheduled" and is not dispatched until scheduled_at is reached. The elected maintenance leader runs a scheduled_to_pending sweep every 1 second, promoting any job whose scheduled_at <= now from scheduled to pending. After promotion the leader sends a pg_notify wake signal so idle workers pick up work immediately.

Timing precision. The sweep fires roughly once per second, so a job may be promoted up to ~1 second after its scheduled_at. Always pass UTC-aware datetimes (datetime.now(UTC)); naive datetimes are not accepted at the backend boundary.

Contents¶

Job lifecycle
JobsClient
enqueue()
enqueue_batch()
enqueue_batch_fast()
Batch failure policies
Batch finalizer
enqueue_batch_streaming()
wait_for_batch()
list_batches()
BatchSummary
JobHandle[R]
get()
get_row()
cancel()
cancel_where()
list()
SubJobEnqueuer
Error handling
Full enqueue-and-wait example
Idempotency example
Batch enqueue example
Tags

`JobsClient`¶

from taskq import TaskQ
from taskq.settings import TaskQSettings

settings = TaskQSettings.load()
async with TaskQ(dsn=str(settings.pg_dsn)) as tq:
    ...  # tq is a JobsClient-compatible client

JobsClient is lightweight — it performs no I/O at construction. Create one instance per application and share it for the lifetime of the process. The connection pool is owned by the Backend, not the client. Creating a JobsClient per-request adds unnecessary overhead and does not provide isolation benefits.

Constructor¶

JobsClient(
    backend: Backend,
    *,
    clock: Clock | None = None,
    settings: TaskQSettings | None = None,
)

Parameter	Type	Default	Description
`backend`	`Backend`	required	The backend to delegate to. In production this is a `PostgresBackend`. In tests, use `InMemoryBackend`.
`clock`	`Clock \\| None`	`SystemClock()`	Retained for API compatibility; it stamps nothing on the enqueue path — "immediate" jobs pass `scheduled_at=None` and the backend's server stamps and decides it. Inject a `FakeClock` in tests for the in-memory backend's own sweep/time-travel semantics.
`settings`	`TaskQSettings \\| None`	`None`	Settings instance threaded through to `JobHandle` for features (e.g. Redis-backed progress fanout) that need config beyond the backend connection.

`backend` property¶

@property
def backend(self) -> Backend: ...

Returns the injected backend. Exposed so JobHandle and tooling can read the backend without accessing the private _backend attribute.

`enqueue()`¶

async def enqueue(
    self,
    ref: ActorRef[P, R],
    payload: P,
    *,
    queue: QueueName | None = None,
    scheduled_at: datetime | None = None,
    priority: int | None = None,
    schedule_to_close: datetime | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
    identity_key: IdentityKey | None = None,
    fairness_key: str | None = None,
    idempotency_key: IdempotencyKey | None = None,
    idempotency_scope: str | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    metadata: dict[str, object] | None = None,
    tags: list[str] | None = None,
) -> JobHandle[R]: ...

Serialises the payload through ref.payload_type, enqueues the job, and returns a typed JobHandle[R].

Parameters¶

Parameter	Type	Default	Description
`ref`	`ActorRef[P, R]`	required	The actor to dispatch.
`payload`	`P`	required	The payload model instance. Re-validated through `ref.payload_type` before insertion.
`queue`	`QueueName \\| None`	`ref.queue`	Override the actor's default queue. Must match `[A-Za-z0-9_][A-Za-z0-9_.-]*` — note `:` is excluded, because it separates segments in the `taskq:global:queue:` concurrency-cap namespace.
`scheduled_at`	`datetime \\| None`	`None`	When to make the job eligible for dispatch. `None` means immediate — the backend's server stamps `scheduled_at` and decides `pending`/`scheduled` (the single-arbiter rule; immune to app↔DB clock skew). Pass a future timezone-aware `datetime` for deferred execution; naive datetimes raise `ValueError`.
`priority`	`int \\| None`	`None`	Dispatch priority. Higher values are dispatched first within the same queue.
`schedule_to_close`	`datetime \\| None`	derived from `retry.time_budget`	Deprecated (emits `DeprecationWarning`): an absolute datetime crosses clock domains (the app clock that produced it vs the database clock that evaluates it) and can misbehave under skew. Declare `retry.time_budget` on the actor instead — the interval form is anchored to the database clock. When supplied (timezone-aware; naive raises `ValueError`) it overrides the `time_budget`-derived interval. Hard deadline: if the job has not reached a terminal state by this datetime it fails with `DeadlineExceeded`.
`start_to_close`	`timedelta \\| None`	`None`	Per-attempt execution timeout measured from when the worker locks the job, enforced via `asyncio.wait_for` around the actor invocation. Distinct from `schedule_to_close` — see `start_to_close` vs `schedule_to_close` for the precedence chain and full explanation.
`heartbeat_timeout`	`timedelta \\| None`	`None`	Maximum time allowed between heartbeats before the job is considered crashed.
`identity_key`	`IdentityKey \\| None`	`None`	Opaque string identifying the logical entity this job belongs to (e.g. `"account:42"`). Required for `unique_for` deduplication to take effect. Also used for fairness scheduling.
`fairness_key`	`str \\| None`	`None`	Partitions the dispatch order so no single key monopolises the queue. Requires the target queue to be in `round_robin` mode (`taskq queues set-mode <queue> round_robin`); on the default `strict_fifo` the key is stored and ignored. See workers.md.
`idempotency_key`	`IdempotencyKey \\| None`	`None`	String preventing duplicate insertion, unique within its `idempotency_scope`. See Idempotency key.
`idempotency_scope`	`str \\| None`	`None`	Namespacing scope for `idempotency_key`. `None` or `""` means the global/default scope (preserves prior global-dedupe behavior). An explicit scope (e.g. a run/batch/epoch id) allows the same business key in different scopes to both succeed. ≤ `idempotency_key_max_bytes` (default 1024 UTF-8 bytes).
`trace_id`	`str \\| None`	extracted from OTel span	Trace ID for distributed tracing. Automatically extracted from the active OTel span when one is valid; pass explicitly to override.
`span_id`	`str \\| None`	extracted from OTel span	Span ID for distributed tracing. See `trace_id`.
`metadata`	`dict[str, object] \\| None`	`{}`	Per-job metadata stored in the `jobs.metadata` JSONB column. Merged with the library-injected `singleton` key when applicable. The caller's dict is never mutated.
`tags`	`list[str] \\| None`	`[]`	Per-job tags stored in `jobs.tags text[]`. Must match `\A\w(?:[\w\-]*\w)?\Z` (1–255 chars). Used for filtering and categorization in queries and the admin UI. See Tags.

Enqueue evaluation order¶

Each call to enqueue() runs these checks in order. A match at step 1 short-circuits all remaining steps. Later steps only execute when earlier ones did not match or raise.

Payload validation — Pydantic re-validates the payload through ref.payload_type. Raises PayloadValidationError on failure (non-retryable).
unique_for dedup check — if identity_key is provided and the actor has unique_for set, scans for an existing job with the same (actor, identity_key) within the window and the configured unique_states. On match, returns the existing handle with was_existing=True and skips all remaining steps.
Singleton pre-flight — if ref.singleton is True, checks for an existing active job for this actor. Raises SingletonCollisionError on collision.
max_pending count check — if ref.max_pending is set, counts pending + scheduled jobs for this actor. Raises MaxPendingExceededError when count >= max_pending.
idempotency_key upsert — if idempotency_key is provided and matches an existing row, returns the existing handle with was_existing=True.
Job INSERT — inserts the new row and returns a handle with was_existing=False.

`idempotency_key`¶

Keys are unique within their idempotency_scope (composite (idempotency_scope, idempotency_key) uniqueness). The default scope (None or "") preserves the prior global-dedupe behavior exactly. Pass an explicit scope (e.g. a run/batch/epoch id) to allow the same business key in different scopes to both succeed, decoupling the dedupe horizon from prune_retention_*. Namespace keys to avoid collisions between actors: "send_receipt:order_123", not "order_123".
Maximum length: 1024 UTF-8 bytes (TASKQ_IDEMPOTENCY_KEY_MAX_BYTES, raisable to 1300). The bound is the composite unique index jobs_idempotency_scope_key_uniq: a Postgres btree v4 entry cannot exceed 2704 bytes, counted encoded — so the cap is in bytes, not characters.
Empty strings and whitespace-only strings raise ValueError before any backend call.
No TTL. idempotency_scope decouples the dedupe horizon by namespace, not by time — there is no idempotency_ttl parameter. A key within a scope still dedupes until pruned, same as before scope existed, just confined to that namespace. This is deliberate: a real sliding-window TTL can't be a single static unique index the way scope can (every mature queue that offers one — Oban, River — trades away the atomic ON CONFLICT insert or buckets time into the key itself). Need "dedupe for an hour, not forever"? Encode the window into the scope string yourself for now.
Rolling-deploy note: mid-upgrade (the pre migration applied, post not yet), reusing a key under two different scopes raises ScopedIdempotencyMigrationPendingError instead of silently deduping against the wrong scope's job — in either direction, so an unscoped call that reuses a key first written under a non-default scope raises it too. Only brand-new keys and same-scope repeats are unaffected. The error message tells the operator to run taskq migrate up --phase post.

Migrating from key-string embedding to scopes. If you previously embedded a run/batch id into the key string to work around global uniqueness, move it to the scope and keep the key as the pure business key:

# Before: run id baked into the key string
await client.enqueue(
    sync_actor,
    payload,
    idempotency_key=f"fetch:{sync_run_id}:{doc.id}",
)

# After: scope carries the run id, key carries the business key
await client.enqueue(
    sync_actor,
    payload,
    idempotency_key=f"fetch:{doc.id}",
    idempotency_scope=sync_run_id,
)

Both forms dedupe within one run; the scoped form additionally lets the same doc.id be re-fetched by a later run without waiting for prune. Note the two forms do not dedupe against each other — a key written as fetch:{run}:{doc} is a different key than fetch:{doc} in scope {run}. Cut over per-actor at a quiet point, or keep the old embedding for in-flight runs and scope only new ones.

idempotency_key does not bypass max_pending. The idempotency check fires at step 5, after the max_pending check at step 4. On the first call with a new key, max_pending is evaluated normally — if the queue is full, MaxPendingExceededError is raised and the job is not inserted. On subsequent calls with the same key (after the key was successfully inserted), the existing JobHandle is returned at step 5 without re-checking max_pending. Only unique_for (step 2) bypasses max_pending unconditionally.

`unique_for` and `singleton` interaction¶

singleton=True and unique_for can coexist on the same actor. They enforce different constraints:

singleton=True blocks a second enqueue as long as any active job for this actor exists, regardless of identity_key.
unique_for + identity_key deduplicates within the configured window per identity.

When both are set and identity_key is provided, unique_for is evaluated first (step 2). A dedup hit returns the existing handle without reaching the singleton pre-flight check at step 3. If the unique_for window has elapsed and a new job is being inserted, the singleton check at step 3 fires and may raise SingletonCollisionError.

`was_existing`¶

JobHandle.was_existing is True when either the unique_for dedup (step 2) or the idempotency_key upsert (step 5) matched an existing job row. Use this instead of comparing created_at timestamps to detect a deduplicated return:

handle = await client.enqueue(my_actor, payload, idempotency_key="order-123")
if handle.was_existing:
    print("job already enqueued, reusing:", handle.job_id)

The same field is set for unique_for dedup:

handle = await client.enqueue(
    sync_account,
    SyncPayload(account_id="acct_123"),
    identity_key="acct_123",
)
if handle.was_existing:
    print("deduplicated within unique_for window:", handle.job_id)

OTel trace propagation¶

When an active OTel span is valid, trace_id and span_id are extracted automatically. Pass trace_id= and span_id= explicitly to override or to propagate an external trace context.

`enqueue_batch()`¶

async def enqueue_batch(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | None" = None,
) -> BatchHandle: ...

Enqueues multiple jobs in a single batched INSERT and returns a BatchHandle containing one JobHandle per item.

All items share a single batch_id UUID written into each job's metadata.batch_id field. Supply batch_id to set it explicitly; omit it and a UUIDv7 is generated automatically.

Parameters¶

Parameter	Type	Default	Description
`items`	`list[EnqueueItem]`	required	1–1000 items to enqueue.
`batch_id`	`UUID \\| None`	auto-generated UUIDv7	Shared identifier for all jobs in this batch.
`connection`	`asyncpg.Connection \\| None`	`None`	Specific connection to use; useful for transactional enqueues.

`EnqueueItem`¶

from taskq import EnqueueItem

EnqueueItem(
    actor_ref=my_actor,  # ActorRef[P, R]
    payload=MyPayload(...),  # validated against actor.payload_type
    scheduled_at=None,  # datetime | None
    priority=None,
    fairness_key=None,
    idempotency_key=None,  # str | None, ≤ 1024 UTF-8 bytes by default
    idempotency_scope=None,  # str | None, ≤ 1024 UTF-8 bytes by default
    identity_key=None,
    metadata={},
)

Field	Type	Default	Description
`actor_ref`	`ActorRef[Any, Any]`	required	The actor to dispatch.
`payload`	`BaseModel`	required	Payload instance; validated against `actor_ref.payload_type` before any INSERT.
`scheduled_at`	`datetime \\| None`	`None`	Deferred execution time.
`priority`	`int \\| None`	`None`	Dispatch priority within the queue.
`fairness_key`	`str \\| None`	`None`	Fairness grouping key. Only affects dispatch on a `round_robin` queue -- see workers.md.
`idempotency_key`	`IdempotencyKey \\| str \\| None`	`None`	Per-item idempotency token (≤ `idempotency_key_max_bytes`, default 1024 UTF-8 bytes).
`idempotency_scope`	`str \\| None`	`None`	Per-item idempotency scope (≤ `idempotency_key_max_bytes`, default 1024 UTF-8 bytes). `None` or `""` = global/default scope.
`identity_key`	`IdentityKey \\| None`	`None`	Opaque identity string; required for `unique_for` dedup to take effect.
`metadata`	`dict[str, object]`	`{}`	Per-job metadata. Do not set `batch_id` here — the library overwrites it.
`tags`	`list[str] \\| None`	`None`	Per-job tags. See Tags.

Deadline anchoring on the batch/COPY arms¶

On every enqueue arm — single, batch, and COPY (enqueue_batch_fast) — an interval-form schedule_to_close (retry.time_budget on the actor) is anchored to the database clock at enqueue time (clock_timestamp() + interval). The batch and COPY arms previously anchored it to scheduled_at + interval, so a future-scheduled item's budget started when it became dispatchable. Under the unified contract a future-scheduled batch item with a short time_budget fails DeadlineExceeded (via the deadline sweep) before it is ever dispatched. If you batch future-scheduled jobs with deadlines, size time_budget from enqueue time, not from scheduled_at.

Validation¶

len(items) == 0 raises ValueError.
len(items) > 1000 raises ValueError.
All payloads are validated before any INSERT. A single validation failure raises PayloadValidationError and leaves no rows inserted.
max_pending is checked in one aggregated query across all actors in the batch. Any violation raises MaxPendingExceededError before the INSERT.
Idempotency-key collisions return the existing JobHandle with was_existing=True, same as single-item enqueue().

`BatchHandle`¶

enqueue_batch() returns a BatchHandle:

Field	Type	Description
`batch_id`	`UUID`	Shared ID for all jobs in this batch.
`job_handles`	`list[JobHandle[Any]]`	One handle per item in the original list.
`size`	`int`	`len(job_handles)`.

`BatchHandle.status()`¶

async def status(
    self,
    db: asyncpg.Connection,
    *,
    schema: str = "taskq",
) -> BatchCompletionStatus: ...

Issues a single GROUP BY query against the jobs table (using the GIN-indexed metadata @> containment filter) and returns aggregated counts:

Field	Type	Description
`total`	`int`	Total jobs in the batch.
`pending`	`int`	Jobs still in flight (`pending + scheduled + running`).
`succeeded`	`int`	Jobs that completed successfully.
`failed`	`int`	Jobs that exhausted retries.
`cancelled`	`int`	Cancelled jobs.
`crashed`	`int`	Jobs that crashed without a clean failure.
`abandoned`	`int`	Jobs abandoned after heartbeat timeout.
`is_complete`	`bool` (computed)	`True` when `pending == 0`.

status = await batch_handle.status(db_connection)
if status.is_complete:
    print(f"batch done — {status.succeeded} succeeded, {status.failed} failed")
else:
    print(f"{status.pending}/{status.total} jobs still running")

BatchHandle.status() vs wait_for_batch(). BatchHandle.status() is a one-shot poll — call it from client-side code (a request handler, a script, a poll loop) whenever you want a snapshot of a batch's completion. taskq.batch.wait_for_batch(db, batch_id) is a different, in-actor helper: call it from inside a finalizer actor holding an asyncpg connection — it raises Snooze(snooze_interval) while jobs are still in flight so the actor's own retry/snooze loop drives the wait, and returns BatchCompletionStatus once all jobs are terminal. Use wait_for_batch() for the fan-out-then-finalize pattern (see batch enqueue example); use BatchHandle.status() everywhere else.

`enqueue_batch_fast()`¶

async def enqueue_batch_fast(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | None" = None,
) -> int: ...

Enqueues jobs via the PG COPY FROM protocol for maximum throughput. Returns the count of inserted rows — no BatchHandle or JobHandle instances.

Tradeoffs vs `enqueue_batch()`¶

Feature	`enqueue_batch()`	`enqueue_batch_fast()`
Protocol	UNNEST INSERT	COPY FROM
Throughput	~10K-50K rows/s	~100K-500K rows/s
Max batch size	1,000	50,000
Idempotency key	Yes (ON CONFLICT)	No (duplicate key aborts entire batch)
Return value	`BatchHandle` with `JobHandle` per item	`int` (row count)
`max_pending` check	Yes	No
Partial success	Yes	No (all-or-nothing atomicity)

Limitations¶

No idempotency-key collision handling. A duplicate key raises asyncpg.UniqueViolationError and aborts the entire batch. Callers must pre-deduplicate.
No max_pending check. The caller is responsible for ensuring actor limits are not exceeded.
No JobHandle instances. Only the inserted count is returned. Use batch_id to query rows post-insert.
All-or-nothing. The COPY fails entirely on any constraint violation — singleton, unique index, or CHECK constraint.

Validation¶

len(items) == 0 raises ValueError.
len(items) > 50_000 raises ValueError.
ALL payloads are validated before any INSERT. A single PayloadValidationError aborts the batch.

Use for bulk import / backfill with 1K–50K rows where throughput matters more than idempotency guarantees.

Batch failure policies¶

A BatchFailurePolicy decides whether a batch should be aborted after observing a run of consecutive job failures. Pass it to enqueue_batch() or enqueue_batch_streaming() via the failure_policy parameter.

from taskq import AbortBatchAfter

batch = await client.enqueue_batch(
    items,
    failure_policy=AbortBatchAfter(consecutive_failures=5),
)

When failure_policy is set, a batches row is created. After each child job reaches a terminal state, the apply_batch_terminal_outcome hook inspects the outcome:

succeeded resets the consecutive-failure counter to 0.
failed increments the counter. If it reaches the threshold, the batch is aborted: all pending and scheduled jobs are cancelled (pending / scheduled → cancelled) and the batch row is set to aborted. Running jobs continue to completion.
cancelled / crashed do not touch the failure counter; they count remaining non-terminal jobs and complete the batch if none remain.

`AbortBatchAfter`¶

from taskq import AbortBatchAfter

AbortBatchAfter(consecutive_failures=5)  # abort after 5 consecutive failures

Parameter	Type	Default	Description
`consecutive_failures`	`int`	required	Abort once consecutive failures reach this count. Must be `>= 1`.

`BatchAbortedError`¶

Raised by wait_for_batch() when the batch row has status = 'aborted' and all child jobs are terminal.

from taskq import BatchAbortedError

try:
    status = await wait_for_batch(db, batch_id)
except BatchAbortedError as exc:
    print(
        f"batch {exc.batch_id} aborted after {exc.consecutive_failures} failures (threshold={exc.threshold})"
    )

Batch finalizer¶

A finalizer is a job enqueued alongside the batch that is dispatched immediately; the in-actor wait_for_batch snooze pattern gates on child-job completion. Pass an EnqueueItem to the finalizer parameter:

from taskq import EnqueueItem, AbortBatchAfter

finalizer = EnqueueItem(
    actor_ref=summarize_results,
    payload=SummarizePayload(batch_id=batch_id),
)

batch = await client.enqueue_batch(
    items,
    failure_policy=AbortBatchAfter(consecutive_failures=5),
    finalizer=finalizer,
)

Transactional enqueue¶

When failure_policy or finalizer is set and no caller-owned connection is provided, the entire operation (batch row + all child jobs + finalizer) is inserted in a single transaction via Backend.enqueue_batch_atomic. If any insert fails, no rows are committed.

When a connection is provided, the caller controls the transaction boundary; the batch row and finalizer are created as the last statements on that connection.

Finalizer is NOT stamped with `batch_id`¶

The finalizer job's metadata does not include batch_id. This is deliberate: if the finalizer were stamped, wait_for_batch would count it as a child, and the finalizer would wait for itself — a deadlock. The batch row's finalizer_job_id column records the link instead, and wait_for_batch automatically excludes that job from counts.

`wait_for_batch` snooze pattern¶

Inside the finalizer actor, call wait_for_batch() to block until all child jobs are terminal. The snooze pattern uses Snooze exceptions so the actor's own retry/snooze loop drives the wait without consuming retry budget:

from taskq import wait_for_batch
from taskq.context import JobContext
import asyncpg


@actor(queue="finalizers")
async def summarize_results(ctx: JobContext, payload: SummarizePayload) -> None:
    db = ctx.deps.worker_pool  # or acquire from pool
    status = await wait_for_batch(
        db,
        payload.batch_id,
        snooze_interval=timedelta(seconds=15),
    )
    # All child jobs are terminal — proceed with finalization
    print(f"batch done: {status.succeeded} succeeded, {status.failed} failed")

wait_for_batch() raises Snooze(snooze_interval) while children are in flight; the consumer transitions the finalizer to scheduled and retries it after snooze_interval without consuming retry budget. Once all children are terminal, it returns BatchCompletionStatus.

`enqueue_batch_streaming()`¶

async def enqueue_batch_streaming(
    self,
    items: Iterable[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | None" = None,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | None = None,
    chunk_size: int = 1000,
) -> BatchHandle: ...

Enqueues jobs from a lazy Iterable[EnqueueItem] (including generators) in chunks of chunk_size (1–1000). All items share the same batch_id. Useful for unbounded iterables where materialising the full list in memory is undesirable.

from taskq import EnqueueItem


async def enqueue_from_generator(client: JobsClient, doc_ids: Iterable[str]) -> None:
    def gen() -> Iterable[EnqueueItem]:
        for doc_id in doc_ids:
            yield EnqueueItem(actor_ref=process_doc, payload=DocPayload(id=doc_id))

    batch = await client.enqueue_batch_streaming(gen(), chunk_size=500)
    print(f"enqueued {batch.size} jobs, batch_id={batch.batch_id}")

Parameter	Type	Default	Description
`items`	`Iterable[EnqueueItem]`	required	Lazy iterable of items. Must not be empty.
`batch_id`	`UUID \\| None`	auto-generated UUIDv7	Shared identifier for all jobs.
`connection`	`asyncpg.Connection \\| None`	`None`	Caller-owned connection for transactional enqueue.
`failure_policy`	`BatchFailurePolicy \\| None`	`None`	Failure policy for the batch.
`finalizer`	`EnqueueItem \\| None`	`None`	Finalizer job to enqueue alongside the batch.
`chunk_size`	`int`	`1000`	Items per INSERT (1–1000).

When failure_policy or finalizer is set and connection is None, the entire operation is delegated to Backend.enqueue_batch_atomic for single-transaction atomicity. Otherwise, chunks are inserted via Backend.enqueue_batch on the caller-owned connection, with the batch row and finalizer created as the last statements.

`wait_for_batch()`¶

from taskq import wait_for_batch


async def wait_for_batch(
    db: "asyncpg.Connection | asyncpg.Pool",
    batch_id: UUID,
    *,
    schema: str = "taskq",
    snooze_interval: timedelta = timedelta(seconds=10),
    snooze_via_exception: bool = True,
    expect_at_least: int | None = None,
    on_empty: Literal["error", "ok"] = "error",
    exclude_job_id: UUID | None = None,
) -> BatchCompletionStatus: ...

In-actor helper for the fan-out-then-finalize pattern. Queries batch children by batch_id using the GIN-indexed WHERE metadata @> $1::jsonb predicate.

Parameter	Type	Default	Description
`db`	`Connection \\| Pool`	required	asyncpg connection or pool.
`batch_id`	`UUID`	required	Batch to wait on.
`schema`	`str`	`"taskq"`	PG schema name.
`snooze_interval`	`timedelta`	`10s`	Snooze/sleep interval (clamped to minimum 1s).
`snooze_via_exception`	`bool`	`True`	`True` raises `Snooze` (in-actor); `False` blocks via `asyncio.sleep` (scripts/tests).
`expect_at_least`	`int \\| None`	`None`	Raise `EmptyBatchError` if fewer than this many jobs exist and none are in flight.
`on_empty`	`"error" \\| "ok"`	`"error"`	Behaviour when zero jobs and no `batches` row: `"error"` raises `EmptyBatchError`; `"ok"` returns empty status.
`exclude_job_id`	`UUID \\| None`	`None`	Omit a specific job from counts. Defaults to the batch row's `finalizer_job_id`.

`expect_at_least` and `on_empty`¶

expect_at_least validates that the batch has at least the expected number of jobs. If fewer jobs are present and none are in flight, EmptyBatchError is raised:

status = await wait_for_batch(
    db,
    batch_id,
    expect_at_least=100,  # raise if fewer than 100 jobs
)

on_empty controls the behaviour when the batch has zero jobs and no batches row exists (e.g. the batch was never created, or all jobs were pruned):

status = await wait_for_batch(
    db,
    batch_id,
    on_empty="ok",  # return empty status instead of raising
)

See the architecture reference for the full decision table.

`list_batches()`¶

from taskq import BatchFilter


async def list_batches(
    self,
    filter: BatchFilter,
) -> list[BatchSummary]: ...

Lists batches matching filter, returning BatchSummary objects with live job counts.

`BatchFilter`¶

from taskq import BatchFilter

BatchFilter(
    queue="notifications",  # str | None
    active=True,  # bool | None — True: active, False: terminal
    batch_id=UUID(...),  # UUID | None
    limit=50,  # int (>= 0, default 100)
)

Field	Type	Default	Description
`queue`	`str \\| None`	`None`	Filter by queue name.
`active`	`bool \\| None`	`None`	`True` selects `active` batches; `False` selects `complete`/`aborted`.
`batch_id`	`UUID \\| None`	`None`	Filter by batch ID.
`limit`	`int`	`100`	Maximum results (must be `>= 0`).

BatchFilter is distinct from JobFilter — it carries only batch-relevant fields. Using JobFilter for batch queries would silently ignore job-oriented fields like status, actor, and tags.

`BatchSummary`¶

from taskq import BatchSummary

Field	Type	Description
`batch_id`	`UUID`	Batch ID.
`queue`	`str`	Queue the batch was enqueued on.
`status`	`"active" \\| "complete" \\| "aborted"`	Batch lifecycle status.
`expected_size`	`int`	Number of child jobs enqueued.
`consecutive_failures`	`int`	Current consecutive failure count.
`failure_threshold`	`int \\| None`	Threshold from `AbortBatchAfter`, or `None`.
`finalizer_job_id`	`UUID \\| None`	Finalizer job ID, if one was enqueued.
`originating_actor`	`str \\| None`	Reserved for future use — currently always `None`. Will be populated from the actor context in a future release.
`created_at`	`datetime`	Batch creation time.
`completed_at`	`datetime \\| None`	When the batch reached a terminal status.
`completion`	`BatchCompletionStatus`	Live job counts (see `BatchCompletionStatus`).

`JobHandle[R]`¶

Returned by enqueue(), enqueue_batch(), and get(). The type parameter R flows from the actor's declared return type.

handle: JobHandle[OrderResult] = await client.enqueue(process_order, payload)

Properties¶

Property	Type	Description
`job_id`	`JobId` (UUID)	The job's unique identifier.
`actor_name`	`str`	The actor this job targets.
`queue`	`str`	The queue the job was enqueued on.
`row`	`JobRow`	The last row this handle observed — seeded at construction by the row `enqueue()`/`get()` fetched, advanced by every row fetch through the handle (`refresh()`, `status()`, `wait()`), and free to read (no backend round trip). `progress_stream()` does not advance it — see `progress_stream()`. Read-only; assignment raises `AttributeError`.
`was_existing`	`bool`	`True` when the handle wraps a deduplicated (existing) job rather than a fresh insert.

Methods¶

`wait()`¶

async def wait(self, *, timeout: float | None = None) -> R: ...

Polls the backend at 0.5 s intervals until the job reaches a terminal status, then validates the stored result through result_adapter and returns R. Each poll advances the handle's row property to the fetched row — on return, row is the terminal row the result was extracted from.

Parameter	Type	Default	Description
`timeout`	`float \\| None`	`None`	Maximum seconds to wait. `None` means wait indefinitely.

Returns: R — the actor's validated return value. Never R | None.

Raises:

Exception	When
`JobFailed`	Terminal status is not `"succeeded"` (`"failed"`, `"cancelled"`, `"crashed"`, `"abandoned"`). The `row` attribute carries the full `JobRow` for inspection.
`ResultUnavailable`	Status is `"succeeded"` but no result is stored (TTL expired, actor returned `None` while `R` is non-`None`). The `row` attribute is available.
`TimeoutError`	`timeout` elapsed before a terminal transition was observed.

`status()`¶

async def status(self) -> JobStatus: ...

Single non-blocking backend read returning the current JobStatus. Does not poll. Advances the handle's row property to the fetched row. Raises RuntimeError if the handle was created via ctx.jobs.enqueue() (no client available).

`refresh()`¶

async def refresh(self) -> JobRow: ...

Re-reads the full JobRow from the backend. Returns the current row regardless of status — does not block on terminal state. Raises RuntimeError without a client.

The re-read also advances the handle's row property: after a refresh(), handle.row and the return value are the same row, so a long-lived handle's row stays current as its owner keeps refreshing. For a one-shot fresh read, handle = await client.get(job_id) then handle.row needs no second fetch — get() already read the row.

`attempts()`¶

async def attempts(self) -> list[AttemptRow]: ...

Returns all AttemptRow records for this job, ordered by attempt number. Raises RuntimeError without a client.

`cancel()`¶

async def cancel(self, reason: str | None = None) -> CancelResult: ...

Delegates to JobsClient.cancel(). Raises RuntimeError without a client.

`progress_stream()`¶

async def progress_stream(self) -> AsyncIterator[ProgressEvent]: ...

Streams live progress events for this job. When Redis is configured, subscribes to the per-job Redis pub/sub channel and yields ProgressEvent objects in real time. When Redis is not available, falls back to polling Postgres every 500 ms and synthesising events from row diffs.

Yields until a terminal=True event is produced (the job reached a terminal status).

progress_stream() does not advance the handle's row property — its Redis path fetches no rows, and advancing only on the PG fallback would make the semantics backend-dependent.

ProgressEvent fields:

Field	Type	Description
`job_id`	`UUID`	The job this event belongs to.
`actor`	`str`	Actor name.
`ts`	`datetime`	Server-side timestamp.
`seq`	`int`	Strictly-monotone sequence number. Use for dedup and `Last-Event-ID` resumption.
`status`	`str`	Current job status at publish time.
`step`	`int \\| None`	Step counter, if reported by the actor.
`percent`	`float \\| None`	Completion percentage, if reported.
`detail`	`str \\| None`	Human-readable status message, if reported.
`data`	`dict[str, object] \\| None`	Custom progress data, if reported.
`terminal`	`bool`	`True` when the job has reached a terminal state.

async for event in handle.progress_stream():
    if event.percent is not None:
        print(f"{event.percent:.0f}% — {event.detail}")
    if event.terminal:
        print(f"job finished with status: {event.status}")
        break

Limitations: - Raises NotImplementedError when using InMemoryBackend — the in-memory backend does not support pub/sub. - Requires the redis extra (uv add "taskq-py[redis]") for real-time delivery. Without Redis the fallback polls Postgres at 500 ms intervals.

For the HTTP SSE endpoint that browser clients can subscribe to, see Progress & Streaming.

`get()`¶

async def get(
    self,
    job_id: JobId,
    *,
    result_adapter: TypeAdapter[R] | None = None,
) -> JobHandle[R] | None: ...

Look up a job by ID. Returns None when the job does not exist. result_adapter is optional because a lookup by ID does not carry actor identity; when omitted it defaults to TypeAdapter(type(None)), which is suitable for status-only lookups. Typical sources:

# When you know the actor:
handle = await client.get(job_id, result_adapter=process_order.result_adapter)

# When you only need row metadata (e.g. status, timestamps):
from pydantic import TypeAdapter

handle = await client.get(job_id, result_adapter=TypeAdapter(type(None)))

`get_row()`¶

async def get_row(self, job_id: JobId) -> JobRow | None: ...

Looks up a job by ID and returns the raw JobRow — no JobHandle, no result adapter. Mirrors get()'s contract: a single backend read, None when the job does not exist. This is the direct form for callers that never need the typed-result machinery:

row = await client.get_row(job_id)
if row is not None:
    print(row.status, row.attempt, row.error_class)

When you do want a handle, get() plus the handle's row property is still one round trip — get() already fetched the row, and handle.row exposes it without re-reading:

handle = await client.get(job_id)
if handle is not None:
    row = handle.row  # the row get() just fetched — no second backend read

`cancel()`¶

async def cancel(
    self,
    job_id: JobId,
    reason: str | None = None,
) -> CancelResult: ...

Requests cancellation of a job and returns a CancelResult.

Semantics:

Reads the current row. Raises KeyError if the job does not exist.
Calls backend.write_cancel_request(job_id, reason).
Reads the row again to capture the new status.

previous_status reflects the row at step 1, not atomically at write time (TOCTOU).

`CancelResult`¶

Frozen Pydantic model returned by cancel().

Field	Type	Description
`job_id`	`UUID`	The job that was targeted.
`previous_status`	`JobStatus`	Status at the first read (before the cancel write).
`new_status`	`JobStatus`	Status after the cancel write.
`cancellation_initiated`	`bool`	`True` when the cancel write transitioned the job to `"cancelled"`. `False` when the job was already in a terminal state and no transition occurred.

result = await client.cancel(handle.job_id, reason="user_requested")
if result.cancellation_initiated:
    print(f"job {result.job_id} cancelled (was {result.previous_status})")
else:
    print(f"job already terminal: {result.new_status}")

`cancel_where()`¶

async def cancel_where(
    self,
    filter: JobFilter,
    reason: str | None = None,
    *,
    allow_empty_filter: bool = False,
) -> BulkCancelResult: ...

Cancel all jobs matching filter in a single set-based operation. Pending/scheduled jobs go straight to terminal cancelled; running jobs get cancel_phase=1 (cooperative cancel) — the worker's heartbeat observes the phase change and sets the in-process cancel_event at the next tick.

Guardrail: a filter with no predicates (no queue, status, actor, identity_key, batch_id, tags, or active) is rejected with EmptyFilterError unless allow_empty_filter=True is passed. This prevents accidental full-table cancels.

Filter fields used: queue, status, actor, identity_key, batch_id, tags, active. The limit, cursor, and order_by fields are ignored — a bulk cancel is not paginated.

Snapshot boundary: jobs matching the filter that are enqueued after the statement's snapshot escape this call. Stop producers first, or issue a follow-up call; the returned counts make non-convergence detectable.

result = await client.cancel_where(
    JobFilter(tags=("tenant-acme",), active=True),
    reason="tenant offboarded",
)
print(
    f"cancelled {result.cancelled_directly} pending, "
    f"requested cancel for {result.cancel_requested} running"
)

`BulkCancelResult`¶

Frozen Pydantic model returned by cancel_where().

Field	Type	Description
`cancelled_directly`	`int`	Pending/scheduled jobs moved straight to terminal `cancelled`.
`cancel_requested`	`int`	Running jobs with `cancel_phase=1` set (cooperative cancel).
`cancelled_ids`	`list[UUID]`	IDs of jobs cancelled directly.
`cancel_requested_ids`	`list[UUID]`	IDs of running jobs with cancel requested.
`total_affected`	`int` (property)	`cancelled_directly + cancel_requested`.

For tenant-scale cancels (10^5+ matching rows), partition via filter (e.g. JobFilter(queue=..., tags=...) to split by queue). A single transaction covering 10^6 rows would hold locks too long. The counts let the caller verify completeness and issue follow-up calls for remaining partitions.

`list()`¶

async def list(self, filter: JobFilter) -> JobPage: ...

Lists jobs matching the filter. Returns a JobPage.

`JobFilter`¶

Frozen dataclass. All fields are optional.

Field	Type	Default	Description
`queue`	`str \\| None`	`None`	Filter by queue name.
`status`	`JobStatus \\| Sequence[JobStatus] \\| None`	`None`	Filter by current status — a single status (`"pending"`) or a sequence (`["pending", "running"]`). An empty sequence matches no jobs; unknown values raise `ValueError`. Mutually exclusive with `active`.
`active`	`bool \\| None`	`None`	Meta-filter by terminality: `True` selects non-terminal statuses (pending, scheduled, running), `False` selects terminal ones. Not Celery's 'active' (currently-executing only) — here it means 'not yet finished'. Mutually exclusive with `status`.
`actor`	`str \\| None`	`None`	Filter by actor name.
`identity_key`	`IdentityKey \\| None`	`None`	Filter by identity key.
`batch_id`	`UUID \\| None`	`None`	Filter by batch ID.
`tags`	`tuple[str, ...] \\| None`	`None`	Filter by tags. Uses `&&` (array overlap) with a GIN index. Returns jobs that match any of the given tags.
`order_by`	`JobSortField \\| None`	`None`	Sort order for results. `None` resolves to `JobSortField.SCHEDULED_AT_ASC` — see JobSortField.
`limit`	`int`	`100`	Maximum number of rows to return.
`cursor`	`str \\| None`	`None`	Opaque keyset-pagination token from `JobPage.next_cursor`.

# "Everything still in flight" — pending + scheduled + running:
page = await client.list(JobFilter(queue="payments", active=True, limit=50))

# A specific set of statuses:
page = await client.list(JobFilter(status=["pending", "running"], limit=50))

`JobSortField`¶

Enum controlling the sort order of list() results. Import from taskq.backend._protocol:

from taskq.backend._protocol import JobSortField

Member	Sort order	Use case
`JobSortField.SCHEDULED_AT_ASC`	Earliest `scheduled_at` first	Default (`None` resolves to this). FIFO / queue-depth inspection.
`JobSortField.CREATED_AT_DESC`	Latest `created_at` first	"Most recently enqueued" queries.
`JobSortField.FINISHED_AT_DESC`	Latest `finished_at` first	"Latest completed run" queries. Jobs with `finished_at IS NULL` sort last (`NULLS LAST`).

Cursors are per-ordering

Every ordering pages with a cursor, but a cursor encodes the sort columns of the ordering that produced it — (priority, scheduled_at, id) for the default, (created_at, id) and (finished_at, id) for the others — so cursors are not interchangeable between orderings. JobsClient.list() fills in JobPage.next_cursor for every ordering, encoded from that ordering's own columns, so feed a page's cursor back with the same order_by you listed with. Changing order_by while reusing a cursor is a caller error, not a supported way to re-sort.

Each ordering tie-breaks on id in the same direction as its primary column (created_at DESC, id DESC). That is what makes a page seam expressible as a single keyset comparison; id is UUIDv7 and therefore time-ordered, so it reorders nothing in practice.

Querying the latest run by `identity_key`¶

Combine identity_key filtering with FINISHED_AT_DESC to find the most recent completed run of a logical entity:

from taskq.backend._protocol import JobFilter, JobSortField

page = await client.list(
    JobFilter(
        actor="sync_tenant",
        identity_key="tenant:acme",
        order_by=JobSortField.FINISHED_AT_DESC,
        limit=1,
    )
)
if page.jobs:
    latest = page.jobs[0]
    print(f"last run: {latest.status} at {latest.finished_at}")

For "most recently enqueued" (regardless of completion), use CREATED_AT_DESC:

page = await client.list(
    JobFilter(
        actor="sync_tenant",
        identity_key="tenant:acme",
        order_by=JobSortField.CREATED_AT_DESC,
        limit=1,
    )
)

`JobPage`¶

Frozen dataclass.

Field	Type	Description
`jobs`	`list[JobRow]`	The matched job rows.
`next_cursor`	`str \\| None`	Pagination token for the next page. `None` when no more rows exist.

page = await client.list(JobFilter(queue="payments", status="pending", limit=50))
for job in page.jobs:
    print(job.id, job.actor, job.status)

if page.next_cursor:
    page2 = await client.list(JobFilter(queue="payments", limit=50, cursor=page.next_cursor))

`SubJobEnqueuer`¶

SubJobEnqueuer is accessed as ctx.jobs inside an actor body. It is not instantiated directly by application code. For actor-side usage see Actor API — Sub-job enqueuing.

Handle limitations¶

Handles returned by ctx.jobs.enqueue() do not have a client bound to them. Calling .status(), .refresh(), .attempts(), or .cancel() on these handles raises RuntimeError. .wait() works because it reads through the backend directly. To poll a sub-job's result from outside the actor body, pass its job_id to a full JobsClient instance:

sub_handle = await ctx.jobs.enqueue(process_item, ItemPayload(item_id=item_id))
sub_job_id = sub_handle.job_id  # safe — job_id is always available

# Later, from application code with a full client:
result_handle = await client.get(sub_job_id, result_adapter=process_item.result_adapter)

`enqueue()`¶

async def enqueue(
    self,
    actor_ref: ActorRef[P, R],
    payload: P,
    *,
    connection: asyncpg.Connection | None = None,
    scheduled_at: datetime | None = None,
    priority: int | None = None,
    fairness_key: str | None = None,
    metadata: dict[str, object] | None = None,
    identity_key: IdentityKey | None = None,
    idempotency_key: IdempotencyKey | str | None = None,
    idempotency_scope: str | None = None,
    unique_for: timedelta | None = None,
    unique_states: tuple[JobStatus, ...] | None = None,
    max_pending: int | None = None,
    tags: list[str] | None = None,
    inherit_tags: bool = True,
    schedule_to_close: datetime | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
) -> JobHandle[R]: ...

Enqueues a single sub-job. Accepts the same options as JobsClient.enqueue() except:

No queue override (sub-jobs use the actor's declared queue).
No explicit trace_id / span_id (extracted from the active OTel span).
connection may be passed to use a specific asyncpg.Connection rather than the LOOP-scope connection.

Tag inheritance¶

Sub-jobs inherit the parent job's tags by default (inherit_tags=True). When no explicit tags are passed, the sub-job carries the parent's tags. When explicit tags are passed, they merge with the parent's tags (parent first, deduplicated).

`inherit_tags`	`tags`	Resulting job tags
`True` (default)	`None`	Parent job's tags (or `()` if parent has none)
`True` (default)	`["new-tag"]`	Parent tags + explicit tags, merged (union, parent-first, deduped)
`False`	`None`	`()` (no inheritance)
`False`	`["new-tag"]`	`("new-tag",)` (explicit only)

Pass inherit_tags=False to suppress inheritance for a specific sub-job.

Blast radius: inherited tags make sub-jobs visible to cancel_where filters matching those tags. A shared/utility sub-job enqueued by a tenant-tagged parent will be swept up in that tenant's cancel_where.

enqueue_batch() does not inherit parent tags — batch items carry their own EnqueueItem.tags. This asymmetry is deliberate: batch fan-out callers typically set per-item tags explicitly.

`schedule_to_close` / `start_to_close` / `heartbeat_timeout`¶

schedule_to_close and start_to_close override the actor's declared defaults for this specific sub-job. heartbeat_timeout has no actor-level declaration — the per-call value is the only source. Note that schedule_to_close bounds total wall-clock time including time snoozed on wait_for_batch — finalizer-style sub-jobs that snooze for long periods should set it generously or not at all.

The per-call max_pending= argument is resolved against the operator-owned stored cap and the @actor(...) literal, not in place of them: against a non-NULL stored actor_config.max_pending the tighter of the two wins (min(stored, per_call)) — explicit load shedding is never widened by an operator override, and no code path can raise the operator's cap. With nothing stored, the per-call argument wins outright over the literal (historical behavior — actor code may loosen its own declaration).

`enqueue_batch()`¶

async def enqueue_batch(
    self,
    items: Sequence[EnqueueItem[Any, Any]],
    *,
    batch_id: UUID | None = None,
    connection: asyncpg.Connection | None = None,
) -> list[JobHandle[Any]]: ...

Enqueues multiple sub-jobs. Currently issues N sequential round-trips (single-statement batch INSERT is a future enhancement). Each EnqueueItem carries actor_ref, payload, and the per-job options (scheduled_at, priority, fairness_key, metadata).

All items share a single batch_id UUID written into each job's metadata.batch_id field. When batch_id is omitted a UUIDv7 is auto-generated; pass it explicitly to correlate the sub-jobs with a finalizer job enqueued separately.

from taskq import EnqueueItem

await ctx.jobs.enqueue_batch(
    [
        EnqueueItem(actor_ref=send_email, payload=EmailPayload(to="a@example.com")),
        EnqueueItem(actor_ref=send_email, payload=EmailPayload(to="b@example.com"), priority=1),
    ]
)

Error handling¶

All exceptions are in taskq.exceptions. Import directly:

from taskq.exceptions import (
    MaxPendingExceededError,
    SingletonCollisionError,
    PayloadValidationError,
    JobFailed,
    ResultUnavailable,
)

Exception	Raised when
`MaxPendingExceededError`	`enqueue()` called when `pending + scheduled` count >= `max_pending`. Fields: `actor` (str), `current_count` (int), `max_pending` (int).
`SingletonCollisionError`	`enqueue()` called for a singleton actor that already has an active job. Fields: `actor` (str), `blocking_job_id` (UUID or None), `retry_after` (timedelta or None).
`PayloadValidationError`	Pydantic validation of the payload fails at enqueue time or at dispatch time. Non-retryable regardless of retry policy. Fields: `actor`, `payload_schema_ver`, `validation_errors`.
`JobFailed`	`JobHandle.wait()` observed a non-success terminal status. Field: `row` (JobRow) with `status`, `error_class`, `error_message`, `error_traceback`.
`ResultUnavailable`	`JobHandle.wait()` observed `"succeeded"` but no usable result is stored (TTL expired, `None` returned where `R` is non-`None`). Field: `row` (JobRow).

from taskq.exceptions import JobFailed, ResultUnavailable

try:
    result = await handle.wait(timeout=30.0)
except JobFailed as exc:
    print(f"job {exc.row.id} failed: {exc.row.error_class}: {exc.row.error_message}")
except ResultUnavailable as exc:
    print(f"job {exc.row.id} succeeded but result is gone (TTL?)")
except TimeoutError:
    print("job did not finish within 30 seconds")

Full enqueue-and-wait example¶

import asyncio
from pydantic import BaseModel
from taskq import TaskQ, actor
from taskq.exceptions import JobFailed, MaxPendingExceededError


class TranscribePayload(BaseModel):
    media_url: str
    language: str = "en"


class TranscribeResult(BaseModel):
    transcript: str
    confidence: float


@actor(queue="media", max_pending=500)
async def transcribe_audio(payload: TranscribePayload) -> TranscribeResult:
    # ... call transcription service ...
    return TranscribeResult(transcript="Hello world", confidence=0.98)


async def main() -> None:
    from taskq.settings import TaskQSettings

    settings = TaskQSettings.load()
    async with TaskQ(dsn=str(settings.pg_dsn)) as tq:
        try:
            handle = await tq.enqueue(
                transcribe_audio,
                TranscribePayload(media_url="https://media.example.com/clip.mp3"),
            )
        except MaxPendingExceededError as exc:
            print(f"queue full ({exc.current_count}/{exc.max_pending}), try later")
            return

        print(f"enqueued job {handle.job_id}, was_existing={handle.was_existing}")

        try:
            result: TranscribeResult = await handle.wait(timeout=120.0)
            print(f"transcript: {result.transcript} (confidence {result.confidence:.0%})")
        except JobFailed as exc:
            print(f"failed: {exc.row.error_class}: {exc.row.error_message}")
        except TimeoutError:
            print("timed out waiting for transcript")


asyncio.run(main())

Idempotency example¶

Use idempotency_key to prevent duplicate execution when a caller may retry the enqueue call (e.g. after a network error):

async def send_order_confirmation(client: JobsClient, order_id: str) -> str:
    """Enqueue a confirmation email, safe to call multiple times for the same order."""
    handle = await client.enqueue(
        send_confirmation_email,
        ConfirmationPayload(order_id=order_id),
        # Namespace the key to avoid collisions with other actors.
        idempotency_key=f"send_confirmation_email:{order_id}",
    )
    if handle.was_existing:
        print(f"order {order_id}: email already enqueued, returning existing job")
    return str(handle.job_id)

Rules:

The key is unique within its idempotency_scope (the default scope None/"" preserves the prior global-dedupe behavior). Always namespace it: "actor_name:entity_id".
Maximum length: 1024 UTF-8 bytes (TASKQ_IDEMPOTENCY_KEY_MAX_BYTES, raisable to 1300). The bound is the composite unique index jobs_idempotency_scope_key_uniq: a Postgres btree v4 entry cannot exceed 2704 bytes, counted encoded — so the cap is in bytes, not characters. Empty and whitespace-only keys raise ValueError.
A duplicate key returns a handle with was_existing=True pointing at the original job.
idempotency_key does not bypass max_pending on the first call for a given key. If the queue is full when the key is first used, MaxPendingExceededError is raised and no row is inserted. On subsequent calls with the same key (after the key was successfully inserted), the existing handle is returned without checking max_pending. Only unique_for (evaluated at step 2, before max_pending) bypasses the queue-depth check unconditionally.

Batch enqueue example¶

Enqueue a fan-out of notifications and poll until the whole batch is done:

import asyncio
import asyncpg
from pydantic import BaseModel
from taskq import TaskQ, actor, EnqueueItem
from taskq.exceptions import MaxPendingExceededError


class NotifyPayload(BaseModel):
    user_id: str
    message: str


@actor(queue="notifications", max_pending=10_000)
async def send_notification(payload: NotifyPayload) -> None:
    # ... deliver notification ...
    pass


async def notify_users(tq: TaskQ, user_ids: list[str], message: str) -> None:
    items = [
        EnqueueItem(
            actor_ref=send_notification,
            payload=NotifyPayload(user_id=uid, message=message),
            idempotency_key=f"send_notification:{uid}:{message[:32]}",
        )
        for uid in user_ids
    ]

    try:
        batch = await tq.enqueue_batch(items)
    except MaxPendingExceededError as exc:
        print(f"notification queue full ({exc.current_count}/{exc.max_pending})")
        return

    print(f"enqueued {batch.size} notifications, batch_id={batch.batch_id}")

    # Poll until complete (replace with your preferred polling strategy)
    pool = await asyncpg.create_pool(dsn="postgresql://user:pass@localhost/taskq")
    async with pool.acquire() as conn:
        while True:
            status = await batch.status(conn)
            if status.is_complete:
                break
            print(f"  {status.pending}/{status.total} pending…")
            await asyncio.sleep(2.0)

    print(f"batch done — {status.succeeded} ok, {status.failed} failed")

Tags¶

Tags are user-defined keyword labels stored in jobs.tags text[]. They have no functional behaviour — no routing, no priority, no lifecycle side effects — and are meant entirely as a user-specified construct for grouping, filtering, and categorizing jobs.

Adding tags at enqueue¶

handle = await client.enqueue(
    send_email,
    EmailPayload(to="user@example.com"),
    tags=["notification", "priority-high", "tenant-acme"],
)

Tags can also be set per-item in batch enqueues:

items = [
    EnqueueItem(actor_ref=process, payload=p, tags=["batch-abc", "chunk-1"]),
    EnqueueItem(actor_ref=process, payload=q, tags=["batch-abc", "chunk-2"]),
]
await client.enqueue_batch(items)

Tag validation¶

Tags must match \A\w(?:[\w\-]*\w)?\Z: - At least 1 character - Starts and ends with a word character ([a-zA-Z0-9_]) - Middle can contain word characters or hyphens - Maximum 255 characters per tag - Duplicates are silently removed (first-occurrence order preserved) - Empty or invalid tags raise ValueError at enqueue time

Filtering by tags¶

Use JobFilter.tags with array-overlap semantics (matches jobs that have any of the given tags):

page = await client.list(
    JobFilter(
        actor="send_email",
        status="failed",
        tags=["priority-high", "tenant-acme"],
        limit=50,
    )
)

SQL: WHERE tags && $n::text[] backed by a GIN index. Tags filtering works in the admin UI via ?tags=comma,separated,values.

Jobs & Clients¶

Job lifecycle¶

Contents¶

JobsClient¶

Constructor¶

backend property¶

enqueue()¶

Parameters¶

Enqueue evaluation order¶

idempotency_key¶

unique_for and singleton interaction¶

was_existing¶

OTel trace propagation¶

enqueue_batch()¶

Parameters¶

EnqueueItem¶

Deadline anchoring on the batch/COPY arms¶

Validation¶

BatchHandle¶

BatchHandle.status()¶

enqueue_batch_fast()¶

Tradeoffs vs enqueue_batch()¶

Limitations¶

Validation¶

Batch failure policies¶

AbortBatchAfter¶

BatchAbortedError¶

Batch finalizer¶

Transactional enqueue¶

Finalizer is NOT stamped with batch_id¶

wait_for_batch snooze pattern¶

enqueue_batch_streaming()¶

wait_for_batch()¶

expect_at_least and on_empty¶

list_batches()¶

BatchFilter¶

BatchSummary¶

JobHandle[R]¶

Properties¶

Methods¶

wait()¶

status()¶

refresh()¶

attempts()¶

cancel()¶

progress_stream()¶

get()¶

get_row()¶

cancel()¶

CancelResult¶

cancel_where()¶

BulkCancelResult¶

list()¶

JobFilter¶

JobSortField¶

Querying the latest run by identity_key¶

JobPage¶

SubJobEnqueuer¶

Handle limitations¶

enqueue()¶

Tag inheritance¶

schedule_to_close / start_to_close / heartbeat_timeout¶

enqueue_batch()¶

Error handling¶

Full enqueue-and-wait example¶

Idempotency example¶

Batch enqueue example¶

Tags¶

Adding tags at enqueue¶

Tag validation¶

Filtering by tags¶

`JobsClient`¶

`backend` property¶

`enqueue()`¶

`idempotency_key`¶

`unique_for` and `singleton` interaction¶

`was_existing`¶

`enqueue_batch()`¶

`EnqueueItem`¶

`BatchHandle`¶

`BatchHandle.status()`¶

`enqueue_batch_fast()`¶

Tradeoffs vs `enqueue_batch()`¶

`AbortBatchAfter`¶

`BatchAbortedError`¶

Finalizer is NOT stamped with `batch_id`¶

`wait_for_batch` snooze pattern¶

`enqueue_batch_streaming()`¶

`wait_for_batch()`¶

`expect_at_least` and `on_empty`¶

`list_batches()`¶

`BatchFilter`¶

`BatchSummary`¶

`JobHandle[R]`¶

`wait()`¶

`status()`¶

`refresh()`¶

`attempts()`¶

`cancel()`¶

`progress_stream()`¶

`get()`¶

`get_row()`¶

`cancel()`¶

`CancelResult`¶

`cancel_where()`¶

`BulkCancelResult`¶

`list()`¶

`JobFilter`¶

`JobSortField`¶

Querying the latest run by `identity_key`¶

`JobPage`¶

`SubJobEnqueuer`¶

`enqueue()`¶

`schedule_to_close` / `start_to_close` / `heartbeat_timeout`¶

`enqueue_batch()`¶