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. Also the shape of an interrupted job whose hold has not elapsed (see below).
`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`	An operator's cancel request escalated and the actor did not exit within `cancellation_grace_period + cleanup_grace_period`. Never produced by a worker shutdown; a deploy releases the job back to the fleet instead.

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

Interruptions. A graceful shutdown that reaches a still-running job releases it: the row returns to pending (or scheduled behind the remaining termination budget when the actor never unwound), with the interrupted attempt spent and interrupt_count bumped. The spent attempt is what guarantees a re-run can never share the attempt epoch with the process that was still shutting down; the row's interrupt_count column counts how often this has happened, and each release writes one job_events transition with reason = 'interrupted'. An operator cancel in flight when the deploy lands still wins the row. The mirror image, a crash (SIGKILL, OOM, a lost heartbeat), also spends the attempt and additionally writes a job_attempts audit row; see retries.md §12 for the crash-vs-shutdown accounting.

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
TaskQ()
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()
backpressure()
SubJobEnqueuer
Error handling
Full enqueue-and-wait example
Idempotency example
Batch enqueue example
Tags

`TaskQ()`¶

TaskQ is the top-level client: it owns the connection plumbing (pool, Redis, LISTEN transport) and exposes the job operations (tq.enqueue(...), tq.get(...), tq.stream(...)) plus tq.actors. It supports both the async context-manager pattern and explicit open() / close() for frameworks that manage their own lifecycle. The full parameter reference also renders from the class docstring on the Client API reference; this table consolidates the constructor surface in one place.

Parameter	Type	Default	Description
`dsn`	`str \\| None`	`None`	Postgres DSN string. Mutually exclusive with `pool` and `pool_factory` (exactly one of the three is required).
`pool`	`asyncpg.Pool \\| None`	`None`	An already-open pool. Caller-owned: `close()` will not close it.
`pool_factory`	`PoolFactory \\| None`	`None`	Zero-arg async factory returning an `asyncpg.Pool`, invoked and owned by TaskQ at `open()`; `reload_credentials()` re-invokes it to rotate the pool in place. Pair with `taskq.auth.make_pg_pool_factory` for rotating-credential deployments.
`pg_provider`	`PgCredentialProvider \\| None`	`None`	Sugar for `pool_factory=make_pg_pool_factory(dsn, pg_provider, min_size=..., max_size=...)` — requires `dsn`, mutually exclusive with `pool_factory`.
`schema`	`str \\| None`	`None`	TaskQ schema name. When omitted, resolved the way the worker and CLI resolve it: `TASKQ_SCHEMA_NAME` from the environment / `.env` cascade, final default `"taskq"`. An explicit value always wins.
`min_pool_size`	`int`	`1`	Minimum pool connections. Only used when `dsn` or `pg_provider` is provided.
`max_pool_size`	`int`	`5`	Maximum pool connections. Only used when `dsn` or `pg_provider` is provided.
`redis_url`	`str \\| None`	`None`	Redis URL for real-time progress fanout. TaskQ creates and owns the client; `close()` closes it. Mutually exclusive with `redis_client`.
`redis_client`	`redis.asyncio.Redis \\| None`	`None`	An already-open Redis client. Caller-owned; `close()` will not close it.
`pg_conn_factory`	`ConnFactory \\| None`	`None`	Zero-arg async factory returning an `asyncpg.Connection` for the LISTEN/NOTIFY transport behind `watch_reclaims()`. For deployments with no DSN (e.g. AAD-managed-identity auth) that still want the reclaim wake-up. Mutually exclusive with `listen_conn`.
`listen_conn`	`asyncpg.Connection \\| None`	`None`	A pre-constructed LISTEN connection. Caller-owned; TaskQ does not close it. Share one across callers.
`reload_interval`	`float \\| None`	`None`	Seconds between automatic credential-reload rebuilds of a factory-built pool; the client-side `TASKQ_RELOAD_INTERVAL`. `None` derives the cadence from the credential's lease (a leaseless factory is never rebuilt automatically). Requires `pool_factory`/`pg_provider` and a positive value — a `ValueError` at construction otherwise, never a silent ignore.
`poll_timeout`	`float`	`30.0`	Maximum seconds between transport wakeups before re-fetching job state. On Postgres alone, `stream()` has no wakeups and polls every `min(poll_timeout, 0.5)` seconds (never below 0.1 s, jittered ±20%).
`reclaim_event_visibility_delay`	`timedelta \\| None`	`None` (`taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY`, 2 s)	The margin `watch_reclaims()` assumes between a `job_events` writer's INSERT and its COMMIT. Must match the worker fleet's margin: the correctness depends on writer transaction duration, not reader preference.

TaskQ is also the actor-management surface: tq.actors (an ActorsClient) exposes list() / get() / set_capacity() / deregister(); see Actor deregistration.

`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,
    capacity_cache_ttl: float = 5.0,
)

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.
`capacity_cache_ttl`	`float`	`5.0` (`DEFAULT_CAPACITY_CACHE_TTL`)	TTL in seconds of the per-actor `max_pending` capacity cache: how long a cached headroom reading is trusted before it is re-counted from the database. This is the TTL behind the "operator `max_pending` change takes effect fleet-wide within seconds" behavior.

InMemoryBackend and FakeClock come from the test-support namespace; see the Testing API reference for its full surface.

`invalidate_actor_capacity_cache()`¶

def invalidate_actor_capacity_cache(self) -> None

Drops the cached actor_config.max_pending snapshot immediately, so the next enqueue() re-counts headroom from the backend instead of waiting out the capacity_cache_ttl. Not needed in normal operation — staleness is bounded by the TTL — but it is the escape hatch for ops tooling (and tests) that just changed the table and cannot wait out the TTL, e.g. a config editor that wrote a new max_pending via taskq actor-config set and must see its next enqueue admit against the new cap immediately:

import asyncio
from datetime import UTC, datetime

from pydantic import BaseModel

from taskq import JobsClient, actor
from taskq.testing.clock import FakeClock
from taskq.testing.in_memory import InMemoryBackend


class PingPayload(BaseModel):
    n: int = 0


@actor(max_pending=10)
async def ping(payload: PingPayload) -> None:
    print("ping", payload.n)


async def main() -> None:
    backend = InMemoryBackend(clock=FakeClock(start=datetime.now(UTC)))
    client = JobsClient(backend)

    handle = await client.enqueue(ping, PingPayload())
    print(handle.job_id)

    # The tool just rewrote a stored max_pending; drop the cached snapshot
    # so the next enqueue re-counts headroom immediately instead of within
    # capacity_cache_ttl (5s by default).
    client.invalidate_actor_capacity_cache()
    print("capacity snapshot dropped; next enqueue re-reads the backend")


asyncio.run(main())

The same invalidation is available to any process holding a JobsClient; an operator taskq actor-config set on another pod needs no call from this one — its enqueues pick the change up within the TTL on their own.

`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 for this job's first placement only; retries and every other re-pend route by the actor's current queue assignment; see Queue overrides: first placement only. Must match `[A-Za-z0-9_][A-Za-z0-9_.-]*` (note `:` is excluded; 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`	Per-job holder-liveness promise, enforced by the leader's reclaim sweep: a running job whose holder has been silent (no heartbeat) past this timeout is crash-reclaimed even while its global `TASKQ_LOCK_LEASE` is still valid: the shorter of the two deadlines governs. Must be `> 0` (a zero-or-negative value anchors the deadline in the past and reclaims a healthy job on the first sweep tick). Size it `>= 2x` the fleet's `TASKQ_HEARTBEAT_INTERVAL`: the per-job analogue of the `lock_lease >= 4 x heartbeat_interval` invariant (see workers.md). The reclaim event rides the existing `reason='lock_expired'` outbox channel with `cause='heartbeat_timeout'`.
`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.

Queue overrides: first placement only¶

A queue= override governs where a consumer first claims the job: nothing more. The row is placed on that queue with its label and is dispatched by that queue's consumers for its first attempt. The moment the job is handed back to the fleet by anything other than a fresh enqueue ( a failure retry, an admission-denial or actor snooze, a crash reclaim, a shutdown interrupt, an operator retry_job), it is routed at dispatch by the actor's current stored queue assignment (actor_config.queue), not by the label it still carries.

The label is kept as the audit trail of first placement, so be aware when reading job listings (admin UI, CLI): the queue column shows that label, and after a retry it no longer names the queue that will claim the row. The queue that will claim it is the actor's current assignment: the same routing queue the stranded-jobs detectors and the queues overview group by.

Concretely: enqueue(ref, payload, queue="gpu") for an actor whose home queue is default puts the first attempt in front of gpu consumers, and, if that attempt fails, the retry is claimable by default consumers only. The override is not a durable affinity: it does not follow the job across attempts, and it is not re-applied by any retry path.

This is a deliberate routing contract, not an oversight, and it is what keeps a queue move safe: taskq actor-config move-queue (see ActorConfig sync) rewrites an actor's stored assignment in one operator action, and every re-pended row of that actor follows the new assignment immediately; otherwise the move's left-behind tail (running jobs that re-pend after the flip) would be claimable only by consumers of the queue the operator just retired, and would strand the moment that queue's last consumer goes away. The same mechanism that rescues moved jobs is the one that retires your override after the first attempt.

If you need a job class to run on dedicated consumers for its whole life:

Put the actor on that queue: declare @actor(queue="gpu") (or move it there with taskq actor-config move-queue) and let every placement, first or retried, route there. This is the supported way to express resource affinity.
A per-job queue= override is for steering first placement only (for example landing a one-off job on a warm pool) and must not be relied on to keep a job off the actor's home queue after a retry.

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. The check-then-insert is serialized per (schema, actor, identity_key) by a transaction-scoped advisory lock (keyed taskq:unique_for:<schema>:<actor>:<identity_key>, hashed via hashtextextended), so two producers racing on the same logical entity cannot both insert: the loser's preflight finds the winner's row and returns it as the dedup hit. The lock wait is bounded: 5 s by default, tunable with TASKQ_UNIQUE_FOR_LOCK_TIMEOUT_MS (see the max_pending paragraph below for how pools the TaskQ client builds itself deliver a widened budget): the acquire is two-tier (taskq._advisory.acquire_advisory_xact_lock_bounded). An uncontended producer pays exactly one pg_try_advisory_xact_lock statement; a contended one queues server-side (Postgres' own lock scheduler hands the lock to the next waiter as each holder's transaction ends), with the wait bounded by a lock_timeout set inside a savepoint (and restored before the savepoint releases, so nothing leaks to the caller's later statements), plus a client-side backstop that bounds the network-black-hole case where the server never answers. A producer that exhausts the budget gets UniqueForLockTimeoutError (this enqueue wrote nothing; on a caller-owned transaction durability is the caller's decision) instead of queueing indefinitely behind a same-key stampede or a black-holed holder. That error is deliberately not in the BackpressureError family: nothing about capacity is wrong; the dedup answer for one identity could not be determined in time, but it still records against the taskq.backpressure.errors counter under its own kind='unique_for_lock_timeout' label, so the refusal shows up on an operator's dashboard without tripping alerts keyed on the capacity kinds; alerting meant for capacity pressure must filter on the kind label rather than the raw counter. The correct response is to retry the same enqueue: once the winner's row is visible, the retry typically returns it as a dedup hit (was_existing=True). 0 or less disables the bound entirely (the unbounded queueing behavior), matching Postgres' own lock_timeout = 0 convention.
Singleton pre-flight: if ref.singleton is True, checks for an existing active job for this actor. Raises SingletonCollisionError on collision. Cron fires now carry the same metadata["singleton"] stamp and pre-flight: previously the flag was client-enqueue-only (see Cron Scheduling: singleton and max_pending interaction).
max_pending count check: if ref.max_pending is set, counts pending + scheduled jobs for this actor. Raises MaxPendingExceededError when count >= max_pending. Single enqueues are exact under concurrency: the count and the INSERT are serialized per (schema, actor) by a transaction-scoped advisory lock (the same two-tier mechanism unique_for uses: one try-lock statement when uncontended, a server-side bounded blocking acquire when contended), so concurrent producers cannot slip between the check and the insert. The lock wait is bounded: 5 s by default, tunable with TASKQ_MAX_PENDING_LOCK_TIMEOUT_MS (its unique_for and idempotency_key siblings are TASKQ_UNIQUE_FOR_LOCK_TIMEOUT_MS and TASKQ_IDEMPOTENCY_LOCK_TIMEOUT_MS): a producer that cannot acquire the lock within its budget gets MaxPendingLockTimeoutError instead of queueing behind every other racer, so burst tail latency is capped by the budget rather than growing linearly with the number of producers, while contended racers still queue server-side (draining at holder-release rate, not at a client poll cadence), so realistic bursts are admitted rather than shed. Treat the timeout exactly like a cap rejection: retry later or shed load (both errors sit in the BackpressureError family and both record against the taskq.backpressure.errors counter). 0 or less disables the bound entirely, matching Postgres' own lock_timeout = 0 convention.

On pools the TaskQ client builds itself (the dsn= and pg_provider= constructions), the pool's per-query command_timeout is 5 s at these defaults and each lock budget is delivered clamped to 80% of it: 4 s at the defaults. The margin is what lets the server-side lock_timeout fire, and the typed error reach you, before the pool's client-side timer would end the wait as a bare TimeoutError. Widening a budget past its 5 s default re-derives the pool's bound to fit the widened budget at the same margin, so a larger TASKQ_MAX_PENDING_LOCK_TIMEOUT_MS (or its siblings) takes effect end to end; budgets at or below the defaults change nothing about the pool. The knobs resolve through the same layers the worker reads: process environment, then the .env cascade, with DOTENV_OVERRIDE/DOTENV_READ_DOTFILES/DOTENV_READ_ENVIRON honored, so a widening set only in .env reaches the client pool's bound exactly as it reaches the worker's server-side budgets. 0 or less asks the server to wait indefinitely, which no client-built pool can honor; its per-query bound still applies, so on that construction set a large finite value instead. Caller-supplied pools (pool= / pool_factory=) keep their own command_timeout and budgets are delivered as configured; keep your pool's per-query bound above any budget you widen.

Bulk paths (enqueue_batch, enqueue_batch_fast) and cron suppressions deliberately do not take the lock: their aggregated pre-insert count is exact for the batch it admits but approximate under concurrency: separate bulk connections can still race the count and the insert, and the overshoot is bounded by the number of racing batches (documented, not silent). Crons and single-process producers therefore stay exact in practice.

Operational note: the lock exists only for capped actors on the single-enqueue path, is keyed taskq:max_pending:<schema>:<actor> (hashed via hashtextextended), and is transaction-scoped. On a connection TaskQ opens and commits itself for the enqueue, it shows in pg_locks only for the duration of that contended enqueue. But when the enqueue runs on a caller-owned, already open transaction (the shape ctx.jobs uses for a transactional actor's sub-enqueue), the lock is held until that caller's commit or rollback, not released early. A transactional actor's start_to_close is unbounded by default, so a sub-enqueue to a capped actor early in the actor body can hold the lock for the actor's entire run. Any other producer to that same actor queues behind the whole parent transaction and can hit MaxPendingLockTimeoutError even though the actor is nowhere near its cap. This is deliberate: releasing the lock before the caller's commit would let the count and the insert drift apart across the caller's own uncommitted work, but it means the lock's true duration is the caller transaction's lifetime, not "one contended enqueue," whenever the caller supplies the connection. Actors without max_pending never touch it. In-memory backend note: InMemoryBackend enforces max_pending with a plain in-process count and takes no advisory lock at all, so InMemoryBackend-based tests cannot reproduce or catch a regression of this transactional-holder lock-duration cost; only a Postgres-backed test observes it. 5. idempotency_key upsert: if idempotency_key is provided and matches an existing row of the same actor, returns the existing handle with was_existing=True. A match on another actor's row raises IdempotencyKeyActorMismatchError (see below). 6. 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".
A hit on another actor's job is refused, not returned. Uniqueness is per scope across all actors, so enqueue(send_receipt, key="order-1") after enqueue(refund, key="order-1") finds the refund's row. That is not a dedup of the receipt job, and handing back the refund's handle (whose .result() is the refund's) would be indistinguishable from one, so the enqueue raises IdempotencyKeyActorMismatchError, naming both actors and the existing job id, with nothing enqueued. In enqueue_batch the refusal is all-or-nothing: the whole batch is withdrawn, like a singleton collision, so it can be resubmitted after fixing the key without duplicating its other items. enqueue_batch_fast classifies its COPY abort the same way: a pair held by another actor raises the mismatch error (with existing_job_id=None when the holder was an earlier item of the same batch, whose row never persisted) and only a same-actor pair raises DuplicateIdempotencyKeyError. (Other queue libraries fold the actor or worker into their unique key, so their index itself rejects a cross-actor hit; TaskQ's index cannot include the actor without a migration, so the hit is checked instead.) The same-actor case is unchanged: the existing handle comes back with was_existing=True.
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 cannot be enforced by a single static unique index. TaskQ uses scope to decouple the horizon, preserving the atomic ON CONFLICT insert. 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 max_pending count check (step 4) runs before the idempotency-key upsert (step 5). 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. The same applies to a duplicate key: re-enqueuing with a key that already exists raises MaxPendingExceededError when the actor is at its cap - the cap check fires first, so the dedupe upsert never runs and the existing JobHandle is not returned past a full queue. Only on a queue with headroom does a duplicate key return the existing handle at the upsert. 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,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | 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.
`failure_policy`	`BatchFailurePolicy \\| None`	`None`	Abort the batch after observing a run of consecutive job failures. See Batch failure policies.
`finalizer`	`EnqueueItem \\| None`	`None`	A job enqueued alongside the batch and dispatched immediately; the in-actor `wait_for_batch` snooze pattern gates on child-job completion. See Batch finalizer.

`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={},
    start_to_close=None,  # timedelta | None, per-attempt execution timeout
)

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.
`start_to_close`	`timedelta \\| None`	`None`	Per-item per-attempt execution timeout, same semantics as `enqueue()`'s `start_to_close`.

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 admission is partitioned per actor: the backend counts existing pending + scheduled per actor in one aggregated query, admits every within-cap actor's items, and refuses an over-cap actor's items as a whole group (never partially filled up to the cap). When any actor is refused, the within-cap actors' items are still enqueued and BatchMaxPendingExceededError raises afterwards; it names each refused actor, the refused item indices into items, and the admitted count. Retry only the refused items, or give items idempotency_keys so a whole-batch retry deduplicates. The atomic path (failure_policy/finalizer set, no connection) keeps all-or-nothing: a cap violation rolls back the entire single transaction and raises plain MaxPendingExceededError with nothing committed.
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. When a finalizer was enqueued, the finalizer's handle is appended as the last entry of this list (and also set as `finalizer_handle`).
`size`	`int`	The number of non-finalizer items - `len(job_handles) - (1 if finalizer_handle is not None else 0)`; the finalizer's handle is the last entry of `job_handles` and is excluded from `size`.
`finalizer_handle`	`JobHandle[Any] \\| None`	The finalizer job's handle, or `None` when no finalizer was enqueued.

`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 an operator cancel outlasted the grace periods.
`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: per-actor partition	Yes: per-actor partition
Partial success	Cap refusals partition per actor (typed error after partial write); idempotency collisions return existing handles	Cap refusals partition per actor (typed error after partial write); COPY is all-or-nothing on constraint violations

Limitations¶

No idempotency-key collision handling. A duplicate (idempotency_scope, idempotency_key) pair (repeated within the batch or already stored) aborts the entire batch with DuplicateIdempotencyKeyError (nothing is written; the abort itself is deliberate bulk-import semantics), or with IdempotencyKeyActorMismatchError when the pair's holder belongs to a different actor (see idempotency_key). Callers must pre-deduplicate. One carve-out: during the 01.00.03 pre→post migration window, a key reused across different scopes raises ScopedIdempotencyMigrationPendingError instead, matching the other enqueue paths.
max_pending is enforced with the same per-actor partition as enqueue_batch(): within-cap actors' rows are written, an over-cap actor's items are refused, and BatchMaxPendingExceededError raises after the COPY commits; retry only the refused items (indices on the error), or rely on idempotency keys.
No JobHandle instances. Only the inserted count is returned. Use batch_id to query rows post-insert.
All-or-nothing on constraint violations. The COPY fails entirely on any constraint violation: singleton, unique index, or CHECK constraint. Only cap admission partitions.

See ops.md: Fan-out at scale for the batch-API chooser table and the chunking patterns that avoid these limits.

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 issue a completion attempt, and the batch completes if no members remain. The attempt is arbitrated server-side against the live member set, so a stale or concurrent attempt can delay completion but never complete a batch that still has non-terminal members. All of these writes are best-effort: a crash (or a bounded lock wait) between a job's terminal write and its counter/abort/complete write loses that one write, and the leader's stale-batch sweep reconciles the batch's status; a lost failure increment delays a threshold abort (it fires on the next failure) rather than preventing it.

`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, wait_for_batch

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(payload: SummarizePayload, ctx: JobContext) -> 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.

Size the finalizer's deadline generously

A snooze whose now + delay would land past schedule_to_close fails the job with DeadlineExceeded instead of rescheduling. A finalizer gating on a long-running batch needs either no deadline or (on a kind="indefinite" actor; time_budget is inert on any other kind) a retry.time_budget well above the batch's expected duration plus retries. A default-kind (transient) finalizer that snoozes has neither a deadline nor an attempt ceiling, since snooze bumps max_attempts instead of consuming it.

Pattern guidance (when to use a finalizer vs a cursor chain vs recursive fan-out, and how to key chunk jobs for idempotency) is in ops.md: Fan-out at scale.

`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: with a caller-supplied connection all chunks share that connection's open transaction (one transaction aggregate; the caller owns the boundary); with no connection each chunk is its own pool transaction, so progress commits incrementally, chunk by chunk.

No-connection failure surface: when chunk N fails (cap refusal, payload validation, a driver error), chunks 1..N-1 (plus the refusing chunk's within-cap actors, under the per-actor max_pending partition) are already durably committed, nothing is returned (the BatchHandle is only constructed after the stream drains), and the call raises. A cap refusal raises BatchMaxPendingExceededError with stream-global refused item indices and an admitted_count covering every committed item; items after the refusing chunk were never attempted. A blind retry of the full iterable would duplicate the committed prefix: retry safely by giving items idempotency_keys (a retry deduplicates against the committed rows) or by resuming from the refused items.

`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; `reason` distinguishes `"result_ttl_expired"` (the row carries the past `result_expires_at` stamp; the message names the expiry instant and says only that no result is stored, swept after expiry or the actor stored none, the stamp cannot distinguish the two) from `"not_stored"`.
`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). The first event is the row as it stands when the stream opens — the same contract TaskQ.stream() documents — so a consumer attaching to an already-running job does not starve until the actor's next publish.

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: the total order over the job's whole event stream (state-change events consume it too). 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. On either transport, a database failure lasting at least the transport's failure budget ends the stream with StreamUnavailable — the PG poll transport's fetches, and on the Redis path, the safety-net re-reads behind the subscription (shorter blips retry silently; the last failure is chained as __cause__) — see the Exceptions API reference.

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 using set-based SQL, drained in bounded committed batches. 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: the drain re-runs its driving statement per batch, each under a fresh snapshot, so a matching job enqueued mid-drain is picked up by a later batch. Jobs enqueued after the final batch's snapshot escape this call. Stop producers first, or issue a follow-up call; the returned counts make non-convergence detectable. A mid-operation failure leaves the batches that already committed as durable partial progress, and a re-run continues where it stopped; already-cancelled rows are skipped.

from taskq import JobFilter

result = await client.cancel_where(
    JobFilter(tags=("tenant-acme",), unfinished=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), the drain makes progress batch by batch; no single transaction ever holds locks on more than event_writer_batch_size rows, so a very large match set costs time, not lock-hold. The cancel NOTIFYs to running jobs' workers are sent once, after the drain completes, in one batched statement. Partitioning via filter (e.g. JobFilter(queue=..., tags=...) to split by queue) is no longer required for lock safety, but remains useful when you want per-partition counts or parallel calls. 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 `unfinished`.
`unfinished`	`bool \\| None`	`None`	Meta-filter by terminality: `True` selects non-terminal statuses (pending, scheduled, running), `False` selects terminal ones. This is broader than just currently-executing jobs: it includes all work not yet finished. Mutually exclusive with `status`. (Renamed from `active`; the old name stays accepted as a deprecated alias.)
`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: one page. `JobsClient.list()` rejects a limit above `10_000` (`taskq.backend._protocol.MAX_JOB_LIST_LIMIT`; a page materialises every row it returns, so a larger result set is paged with `cursor`). Ignored by `cancel_where`.
`cursor`	`str \\| None`	`None`	Opaque keyset-pagination token from `JobPage.next_cursor`.

from taskq import JobFilter

# "Everything still in flight": pending + scheduled + running:
page = await client.list(JobFilter(queue="payments", unfinished=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` exactly when no more rows exist: the client looks one row past `limit`, so a last page that happens to fill the limit still ends the walk without an empty trailing page (at the `10_000` ceiling there is no room to look past it, and a full page there may lead to one empty page).

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))

`backpressure()`¶

async def backpressure(
    self,
    queues: Sequence[str],
    *,
    parent_id: UUID | None = None,
) -> BackpressureSnapshot: ...

A read-only typed read of the submit path's backpressure state, mirroring the reader methods (get_row, list) — deliberately not an enqueue kwarg: enqueue is generic JobHandle[R] and no flag may flip the reply type. One call answers, per queue: "is this queue at or past its admission boundary?"

snapshot = await client.backpressure(["payments"])
entry = snapshot.queues["payments"]
if entry.state == "over":
    shed_or_delay()
elif entry.state == "unknown":
    log.warning(entry.reason)  # what could not be seen, never a fake verdict

The three halves of every verdict¶

Field	What it is	Cost
`depth`	Jobs holding a pending slot in this queue (pending + scheduled) — the queue-local, ops view. Reported, not verdict-bearing.	One indexed aggregate per call (`count_pending_jobs_by_queue`, the `count_active_jobs` pattern), all queried queues in one round trip.
`binding_actor` / `admission_load` / `effective_max_pending`	The verdict's basis. The admission cap is per-ACTOR and governs the actor's pending+scheduled count across all queues (`enqueue_max_pending_count`: `WHERE actor = $1`, enforced at every enqueue). The binding actor is the queue's routing actor with the smallest headroom (its own cap minus its own all-queue count); `admission_load` is that actor's count; `effective_max_pending` is its stored cap, read from the client-side `ActorCapacityCache` TTL snapshot (default 5s staleness bound — the same snapshot enqueue admission itself resolves through).	Zero I/O beyond one warm-up refresh per TTL window (shared with the enqueue path) plus one `count_pending_jobs` aggregate for the union of the routing actors.
`children_depth`	The fan-out parent's pending children outside this queue — the exact `parent_id` ledger. Reported, not verdict-bearing: every pending child already counts toward its OWN actor's cap (wherever it sits), which is the number admission enforces — the fan-out pressure reaches the verdict through the routing actors' admission loads, and this field is the parent-attributed view of the same rows.	One more aggregate round trip, only when a parent is in play.

The verdict: state is "over" when any routing actor's own all-queue count is at or over its stored cap — the next enqueue for that actor is likely refused with MaxPendingExceededError — and "ok" only when every routing actor resolves and none is over.

Why the verdict is actor-scoped, not queue-scoped¶

The cap is WHERE actor = $1 — an actor's traffic split across queues (an explicit queue= override, or a queue move) accumulates in ONE count that the queue-local depth cannot see. A queue-local verdict false-OKs exactly there: 9 pending in the assigned queue + 9 via the override read "9 < 10, ok" while the next enqueue is refused at 18 >= 10. The verdict therefore compares what the cap actually governs (the actor's own count), and the queue-local depth stays as the ops view. Pinned: test_differential_actor_split_across_queues_is_seen.

The fan-out half: exact `parent_id` accounting¶

A fan-out parent (a dispatcher enqueuing thousands of children) needs its children's pending depth visible. The ledger is exact: every child enqueue made under a parent's context stamps jobs.parent_id with the parent's job id (a plain column, migration 01.00.23_01 — deliberately no foreign key, so it never serializes child inserts behind the parent row and never blocks a retention purge; a dangling parent_id — parent purged, children pending — is a defined, harmless state the count handles by construction).

Inside a worker, the parent context is already installed (the same contextvar pair that drives tag inheritance, set at worker entry), so an actor body needs no arguments:

@actor()
async def dispatch_orders(payload, ctx) -> None:
    snapshot = await ctx.jobs.backpressure(["order_items"])  # ambient parent = THIS job
    entry = snapshot.queues["order_items"]
    if entry.state == "over":
        await asyncio.sleep(1)  # shed: our own children are saturating the target
    for order in payload.orders:
        await ctx.jobs.enqueue(enrich_item, order)  # child: parent_id stamped

Outside a worker (or for an explicit parent), pass parent_id=. The fan-out pressure reaches the verdict through the routing actors' own counts: every pending child is a row of ITS target actor, and that actor's cap governs its all-queue count — so children piling up on the target queue turn the TARGET queue's verdict over, which is exactly the admission truth (the next child will be refused by the same arithmetic). children_depth is the parent-attributed ledger view of those same rows (pending children of the parent outside the queried queue), reported for observability; children already in the queried queue are in its depth and are never counted twice.

Fail-open: `unknown`, never a fabricated verdict¶

Any half the verdict needs that cannot be seen forces state="unknown" with reason naming it: a sick database (the read fails or exceeds its bounded wait), a backend built before the staged reads existed, no capacity snapshot, a queue no stored assignment routes (the unserved-queue corner), or a routing actor whose cap lives only in the @actor(max_pending=...) literal — code is invisible to another process, so set a stored override (taskq actor-config set --max-pending) to make the read exact. When SOME routing actors resolve and others do not, a resolved actor that is over still reads over (that actor's refusal is certain), but ok is withheld (an unresolved actor may be over — claiming ok would be the false-ok). The read never raises on backend trouble; the one exception is a missing schema, which surfaces as SchemaNotMigratedError — a setup defect is not degraded data (the actor body's ctx.jobs.backpressure() translates it identically).

As-of: the snapshot states its own age¶

The snapshot carries as_of (the UTC instant of the read) and cap_age_seconds (how old the capacity snapshot the caps came from was — computed from the last refresh that SUCCEEDED, so a sustained-failure regime reports the data's real age instead of resetting fresh on every failed retry; bounded by the TTL, default 5s; None when no snapshot exists, then every cap half reads unknown with its reason). The depth half was exact at as_of and advisory from it on; a caller surfacing the number to users can show the staleness story honestly instead of guessing (prefect's "Late" state taught this the hard way: a stale verdict presented as fresh).

Dangling parents: first-class states, never errors¶

parent_id has no foreign key, so nothing in the database distinguishes a parent that never existed (a stale or fabricated id) from one purged by retention from one not yet inserted (children stamped before the parent's row commits). The snapshot makes no attempt to distinguish them either — all three are just "children pending", the count counts children by parent_id and never joins to the parent row, and the verdict is the same well-formed ok/over in all three cases. That is deliberate pricing, not an oversight: an advisory signal measures admission pressure, which is a property of the CHILDREN; telling the three dangling cases apart would cost a join per read and buy a distinction no caller can act on. Oban Pro needed three separate mechanisms (Lifeline, ignore_deleted, preserve_workflows) to mop up this ambiguity because its signals special-case missing parents; this read defines the dangling state as ordinary instead. The one honest posture note: pending children of a purged parent run orphaned (no parent to observe them) — that is a lifecycle question for the caller, not a count question, and the snapshot refuses to fabricate a difference.

Retention is linkage-safe by construction¶

The prune sweep's candidates are TERMINAL rows only (finished_at-bounded); a pending child — of a live parent or a purged one — is never a purge candidate, so a purge pass running mid-workflow cannot strand or corrupt pending children. Pinned in tests/test_backpressure_read.py.

Advisory semantics¶

The read is a check; the enqueue is a later act. The race between them is acceptable by design — the signal predicts admission, it does not reserve it. The cap half carries the capacity cache's TTL staleness bound (default 5s); the depth half is a real count but ages the moment it is taken. Callers needing certainty use enqueue's own typed refusal (MaxPendingExceededError), which stays the single enforcement point.

Measured cost (100k-row hot table, PG 18, min of 5): the depth aggregate serves the whole call from jobs_parent_pending_idx-shaped index walks — a 10k-pending-children fan-out counts in ~2.6 ms; the enqueue COPY path's parented vs unparented rate difference is inside noise (the partial index only touches parent-stamped rows). See scripts/measure_parent_id_cost.py.

`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.

Transactional sub-enqueue: session state, not concurrency, is the constraint

Sub-enqueues join the actor's transaction only when a LOOP-scope asyncpg.Connection is registered in DI. With max_concurrency > 1 the worker opens a dedicated per-slot transaction pool, each job transacts on its own connection, and the actor receives that same slot connection by injection: its own writes join the transaction, and concurrent slots never share a connection, so the transactional path is correct at any concurrency (the startup event transactional_consume_per_slot announces the mode). Two consequences to know: the actor's transactional writes run on the slot connections (fresh direct-DSN connections that never carried the registered LOOP-scope connection's session state (SET ROLE, search_path, an RLS-driving GUC), so writes that must inherit that state belong on a TASKQ_MAX_CONCURRENCY=1 worker, where the transaction and the actor's injected connection keep using the registered connection, and the per-slot pool costs max_concurrency + 1 direct connections, so the single-slot worker is also the minimal-connection-budget shape. Without a LOOP-scope Connection (the default worker, including a LOOP-scope asyncpg.Pool registration, which does not activate the transactional path), ctx.jobs commits each child immediately through the worker pool (autonomous mode; the startup log warns sub_enqueue_autonomous_fallback). See ops.md: Fan-out at scale for the consequences for chaining patterns.

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 and is call-site-only: it sets this sub-job's holder-liveness promise, enforced by the leader's reclaim sweep (a holder silent past it is crash-reclaimed even while the global TASKQ_LOCK_LEASE is still valid). 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),
    ]
)

Partial failure raises PartialBatchError. The autonomous mode (no LOOP-scope connection) enqueues each item independently; items enqueued before the first failure are committed, the rest are not inserted, and the call raises with the house shape for partial batch admission: succeeded_count (items committed by this call), failed_items (a list of (index, exception) pairs into the caller's items list), and total (the original batch size). The connection arm speaks the same contract: a per-actor max_pending partition refusal there is converted to PartialBatchError (within-cap actors' items are inserted first). Retry only the failed indices — a blind whole-batch retry duplicates the committed items — or give items idempotency_keys so the retry dedupes. On a LOOP-scope connection the inserted items ride the parent's transaction: catching the error inside the actor body and returning normally commits them; letting it propagate rolls everything back. See the Exceptions API reference.

Error handling¶

All exceptions are in taskq.exceptions. Import directly:

from taskq.exceptions import (
    BatchMaxPendingExceededError,
    MaxPendingExceededError,
    MaxPendingLockTimeoutError,
    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).
`MaxPendingLockTimeoutError`	`enqueue()` on a capped actor could not acquire the per-`(schema, actor)` serialization advisory lock within its budget (default 5 s): cap check never ran. Two distinct causes: too many concurrent producers racing the same actor, or a single long-running caller transaction holding the lock for its own lifetime (e.g. a transactional actor's `ctx.jobs` sub-enqueue to a capped actor, held until that actor's commit/rollback; see the operational note above). Same `BackpressureError` family as `MaxPendingExceededError`; same response (retry later or shed load; for the transactional-holder cause, also check for a long-running actor sub-enqueueing to this capped actor). Fields: `actor` (str), `timeout_ms` (float).
`UniqueForLockTimeoutError`	`enqueue()` with `unique_for` + `identity_key` could not acquire the per-`(schema, actor, identity_key)` single-flight advisory lock within its budget (default 5 s, `DEFAULT_UNIQUE_FOR_LOCK_TIMEOUT_MS`): the dedup answer for one identity could not be determined in time; nothing was inserted. NOT a `BackpressureError` (no capacity problem), though it is counted in `taskq.backpressure.errors` under `kind="unique_for_lock_timeout"` alongside the capacity kinds. Response: retry the same enqueue: it typically dedupes against the winner's row (`was_existing=True`). Fields: `actor` (str), `identity_key` (str), `timeout_ms` (float).
`BatchMaxPendingExceededError`	A bulk enqueue (`enqueue_batch()` / `enqueue_batch_fast()` / the chunked arm of `enqueue_batch_streaming()`) partitioned admission per actor and refused some: the within-cap actors' items were inserted first, then this raises. Fields: `refusals` (list of `MaxPendingExceededError`, one per over-cap actor), `refused_indices` (actor -> indices into the caller's items), `admitted_count` (int). Not a `MaxPendingExceededError` subclass: part of the batch is already stored; retry only the refused items or rely on idempotency keys. An `except BackpressureError` handler catches this too and must consult `admitted_count` / `refused_indices` before any whole-batch retry.
`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`). Fields: `row` (JobRow), `reason` (`"result_ttl_expired"`

Catching backpressure generically. MaxPendingExceededError, SingletonCollisionError, and BatchMaxPendingExceededError all subclass BackpressureError, so except BackpressureError is a valid catch-all for enqueue-time backpressure, with one hazard: the batch variant raises after the within-cap actors' items are stored, so a generic handler that retries the whole batch duplicates them. Consult admitted_count / refused_indices before retrying (retry only the refused items, or rely on idempotency_keys to deduplicate a whole-batch retry).

For the retention gap signal (EventRetentionGapError, raised by a resumed watch_reclaims stream whose cursor the event-prune watermark passed): recovery after a crash-before-consume is to recreate the watcher fresh (watch_reclaims(after_id=0)), accepting the loss the error reported, the deleted events cannot be refilled.

The catch-all. Every library-raised error subclasses TaskQError — the table above is the typed layer, and except TaskQError is the correct fallback in handlers that must never let a library error escape uncaught. Two deliberate escapees do NOT subclass it: UnencodableValue subclasses TypeError (the historical orjson contract — an except TypeError swallows it) and RateLimitDependencyUnavailable (with its subclass RateLimitStoreCorrupt) subclasses RuntimeError. If those three matter to your handler, catch them explicitly beside the TaskQError clause. The full taxonomy — every class, its fields, and its remedy — is indexed in the Exceptions API reference.

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()

    # First run on a fresh database: apply the schema migrations
    # (`taskq migrate up` does the same from the CLI).
    import asyncpg
    from taskq.migrate import apply_pending_locked

    conn = await asyncpg.connect(str(settings.pg_dsn))
    try:
        await apply_pending_locked(conn=conn, schema=str(settings.schema_name))
    finally:
        await conn.close()

    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=30.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())

The worker must consume the actor's queue

This actor is declared on queue="media"; a worker started without --queues media consumes only default, so handle.wait() would time out with the job parked and nothing to claim it: taskq worker --actors myapp.actors:registry --queues media.

This example is the spine of the fleet demo (examples/fleet_demo/; run with uv run python -m examples.fleet_demo.run_demo): it enqueues through the client, cancels through taskq job cancel, and reads the run back through taskq insights — on real containers, with a real worker subprocess.

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 - the max_pending count check runs before the idempotency-key upsert. If the queue is full, MaxPendingExceededError is raised and no row is inserted, whether the key is new or a duplicate: re-enqueuing with an existing key on an actor already at its cap raises the error rather than returning the existing handle. On a queue with headroom, a duplicate key returns a handle with was_existing=True. 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¶

TaskQ()¶

JobsClient¶

Constructor¶

invalidate_actor_capacity_cache()¶

backend property¶

enqueue()¶

Parameters¶

Queue overrides: first placement only¶

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¶

backpressure()¶

The three halves of every verdict¶

Why the verdict is actor-scoped, not queue-scoped¶

The fan-out half: exact parent_id accounting¶

Fail-open: unknown, never a fabricated verdict¶

As-of: the snapshot states its own age¶

Dangling parents: first-class states, never errors¶

Retention is linkage-safe by construction¶

Advisory semantics¶

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¶

`TaskQ()`¶

`JobsClient`¶

`invalidate_actor_capacity_cache()`¶

`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`¶

`backpressure()`¶

The fan-out half: exact `parent_id` accounting¶

Fail-open: `unknown`, never a fabricated verdict¶

`SubJobEnqueuer`¶

`enqueue()`¶

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

`enqueue_batch()`¶