Package Overview¶
TaskQ public API. The taskq package re-exports the primary types used by application
code: the @actor decorator, JobsClient / TaskQ, JobHandle, exceptions,
RetryPolicy, cron scheduling, and batch helpers.
from taskq import actor, TaskQ, JobHandle, JobFailed, RetryPolicy
from taskq import cron, ScheduleHandle
from taskq.context import JobContext
from taskq.di import ProviderRegistry, Scope
taskq ¶
TaskQ: async-native, Postgres-backed background job library.
Canonical imports:
from taskq import actor, TaskQ, JobHandle, JobFailed, RetryPolicy
from taskq import cron, ScheduleHandle
from taskq.context import JobContext
from taskq.di import ProviderRegistry, Scope
TERMINAL_STATUSES
module-attribute
¶
TERMINAL_STATUSES: frozenset[JobStatus] = frozenset(
{
"succeeded",
"failed",
"cancelled",
"crashed",
"abandoned",
}
)
IdempotencyKey
module-attribute
¶
Distinguishes idempotency keys from identity keys at call sites.
IdentityKey
module-attribute
¶
Distinguishes identity keys from idempotency keys at call sites.
JobId
module-attribute
¶
Opaque job identifier — prevents UUID mixups across the API.
QueueName
module-attribute
¶
Validator alias for queue names — accepts plain str literals.
Why Annotated and not NewType: every studied vendor (river,
dramatiq, arq, procrastinate) uses raw str + a separate validator
for queue names; no nominal type because no other str field at any
call site could be confused with queue. Annotated gives runtime
validation in Pydantic models without forcing every caller to wrap
literals in QueueName("default").
__all__
module-attribute
¶
__all__ = [
"TERMINAL_STATUSES",
"AbortBatchAfter",
"ActorConfigDriftError",
"ActorConfigDriftList",
"ActorConfigRow",
"ActorDeregistrationError",
"ActorFn",
"ActorFnWithCtx",
"ActorHandler",
"ActorHasActiveJobsError",
"ActorHasEnabledSchedulesError",
"ActorNotFoundError",
"ActorRef",
"ActorsClient",
"Backend",
"BackpressureError",
"BatchAbortedError",
"BatchCompletionStatus",
"BatchCounts",
"BatchFailurePolicy",
"BatchFilter",
"BatchHandle",
"BatchRow",
"BatchSummary",
"BulkCancelResult",
"CancelPhase",
"CancelResult",
"Clock",
"ConnFactory",
"CronScheduleSpec",
"DIError",
"DependencyCycle",
"DeregisterResult",
"DstStrategy",
"EmptyBatchError",
"EmptyFilterError",
"EnqueueItem",
"ErrorReporter",
"EventRow",
"Fail",
"FakeClock",
"IdempotencyKey",
"IdentityKey",
"IllegalStateTransition",
"JobContext",
"JobEvent",
"JobFailed",
"JobFilter",
"JobHandle",
"JobId",
"JobPage",
"JobRetryState",
"JobRow",
"JobSortField",
"JobStatus",
"JobsClient",
"MaxPendingExceededError",
"MissingProvider",
"NullErrorReporter",
"OIDCSettings",
"OnSuccess",
"PartialBatchError",
"PayloadValidationError",
"PgCredential",
"PgCredentialProvider",
"PoolFactory",
"ProgressEvent",
"ProgressTooLarge",
"QueueMode",
"QueueName",
"RateLimitBackend",
"RedisCredential",
"RedisCredentialProvider",
"RedisFactory",
"ReservationUnavailable",
"ResultTooLarge",
"ResultUnavailable",
"Retry",
"RetryAfter",
"RetryClassifier",
"RetryClassifierHook",
"RetryDecision",
"RetryKind",
"RetryOverride",
"RetryPolicy",
"SAMLSettings",
"ScheduleHandle",
"ScheduleRecord",
"SchemaNotMigratedError",
"ScopeViolation",
"ScopedIdempotencyMigrationPendingError",
"SingletonCollisionError",
"Snooze",
"SubEnqueueError",
"SubJobEnqueuer",
"SystemClock",
"TaskQ",
"TaskQError",
"TaskQSettings",
"WorkerConnections",
"WorkerOwnershipMismatch",
"WorkerSettings",
"__version__",
"actor",
"apply_batch_terminal_outcome",
"cron",
"enrich_pg_dsn",
"ensure_sslmode_require",
"make_dedicated_conn_factory",
"make_pg_pool_factory",
"make_redis_client_factory",
"register_cron",
"validate_actor_payload",
"wait_for_batch",
]
ActorFn ¶
Actor handler that takes only a payload.
The dispatcher injects nothing beyond the validated payload model. Use this shape for actors that don't need cancellation cooperation, attempt counters, or other context fields.
ActorFnWithCtx ¶
Actor handler that takes a payload and a typed :class:JobContext.
Declare ctx: JobContext[YourPayload] as the second parameter to
opt into context injection. The dispatcher constructs the context per
attempt, populates it with the validated payload and a fresh
:class:asyncio.Event for cooperative cancellation, then passes it
to the handler. Handlers that don't declare ctx skip this work.
JobStatus ¶
JobStatus = Literal[
"pending",
"scheduled",
"running",
"succeeded",
"failed",
"cancelled",
"crashed",
"abandoned",
]
RetryKind ¶
Closed set of retry tiers.
Why Literal and not an Enum: serialization round-trips through
model_dump(mode="json") produce plain strings without
use_enum_values configuration; pyright exhaustive matching works
identically for either; no .value access required at call sites.
OnSuccess ¶
Hook fired when a job succeeds. Receives (job_row, result).
Why object for the result type (not Any or generic R): the
hook is dispatched from the consumer loop, which erases the actor's
return type to object. This mirrors the non-generic
:data:OnRetryExhausted at the same payload-erasure boundary. Hooks
that need a typed result re-validate via the actor's
result_adapter.
RetryClassifierHook ¶
Optional per-actor hook for exception-instance-level retry classification.
non_retryable_exceptions and the built-in :class:PayloadValidationError
check classify by exception type alone. Some integrations need finer
granularity — a single exception type (e.g. an HTTP client's status-code
error) that should retry indefinitely on a 429, fail immediately on a 404,
and use a bounded transient budget on a 5xx, or a server-provided
Retry-After value that should drive the actual backoff delay. Register
one via @actor(retry_classifier=...).
Invoked with (exception, attempt) for every exception that survives the
non_retryable_exceptions/PayloadValidationError checks. Return
None to fall back to the actor's static RetryPolicy unchanged, or a
:class:RetryOverride to refine kind and/or delay for this specific
occurrence. Exceptions raised by the hook itself are caught and logged by
:func:decide_after_failure; classification falls back to the static
policy in that case — a broken hook can never crash the retry pipeline.
ActorHandler ¶
Bases: Protocol
Most-general actor signature: payload first, then ctx and/or DI deps.
Pyright infers P_ from the first positional parameter (the
payload model) and R_ from the awaited return type, regardless
of how many additional ctx / DI parameters the handler
declares. This lets the :func:actor decorator preserve full
payload-and-result inference for FastAPI-style handlers like::
async def my_actor(
payload: OrderPayload,
ctx: JobContext[OrderPayload],
*,
db: DbSession,
http: HttpClient,
) -> OrderResult: ...
Without ActorHandler, callers would have to choose between
:data:ActorFn (payload only) and :data:ActorFnWithCtx (payload
+ ctx), neither of which describes the DI case.
ActorRef ¶
ActorRef(
*,
name: str,
queue: str,
fn: Callable[..., object],
is_sync: bool = False,
wants_ctx: bool,
dependencies: dict[str, type[object]],
payload_type: type[P],
result_adapter: TypeAdapter[R],
retry: RetryPolicy,
result_ttl: timedelta | None,
singleton: bool = False,
max_concurrent: int | None = None,
max_pending: int | None = None,
metadata: dict[str, object] | None = None,
unique_for: timedelta | None = None,
unique_states: tuple[JobStatus, ...] = (
"pending",
"scheduled",
"running",
),
start_to_close: timedelta | None = None,
rate_limits: list[
str
| KeyedRateLimitRef
| TokenBucket
| SlidingWindow
]
| None = None,
reservations: list[
str | KeyedReservationRef | ConcurrencyReservation
]
| None = None,
non_retryable_exceptions: tuple[
type[BaseException], ...
] = (),
retry_classifier: RetryClassifierHook | None = None,
on_retry_exhausted: OnRetryExhausted | None = None,
on_retry_exhausted_timeout: float = 3.0,
on_success: OnSuccess | None = None,
on_success_timeout: float = 3.0,
priority: int = 0,
)
Typed reference to a registered actor.
Created by the :func:actor decorator. Not callable directly via
the queue path — enqueue jobs by passing this ref to
:meth:JobsClient.enqueue(ref, payload) <taskq.client.JobsClient.enqueue>.
Direct in-process invocation (await my_actor(payload, ...)) is
available for tests and simulators.
The two type parameters carry the actor's payload and result types end-to-end:
payload_typere-validates raw dispatch-time payloads back toP(via :meth:pydantic.BaseModel.model_validate).result_adapterround-trips actor returns through the JSONBresultcolumn (dump_python(mode="json")on the worker side,validate_pythonon the client side).
Attributes:
| Name | Type | Description |
|---|---|---|
max_concurrent |
Fleet-wide concurrency cap for this actor.
max_concurrent may transiently exceed configured value by up to (num_active_producers - 1) * max_concurrent per actor under heavy contention (or (num_producers - 1) * limit_n when limit_n < max_concurrent). For strict correctness, use ConcurrencyReservation. |
|
metadata |
Arbitrary key-value metadata stored in
|
Both are stored as instance fields rather than class metadata so
pyright can infer P and R from a constructor call without
relying on phantom-type tricks.
wants_ctx records whether the handler declared a
:class:JobContext parameter; the dispatcher passes ctx only
when set. dependencies maps each DI parameter name to its
annotated type — the worker's DI pass resolves these at dispatch
time and passes them as keyword arguments to the handler. The DI
resolver itself is an erasure boundary (see
the resolver operates on the registered provider graph at runtime): user-declared annotations on
DI parameters are captured here as type[object] because the
resolver operates on the registered provider graph at runtime.
Source code in src/taskq/actor.py
__slots__
class-attribute
instance-attribute
¶
__slots__ = (
"_fn",
"dependencies",
"is_sync",
"max_concurrent",
"max_pending",
"metadata",
"name",
"non_retryable_exceptions",
"on_retry_exhausted",
"on_retry_exhausted_timeout",
"on_success",
"on_success_timeout",
"payload_type",
"priority",
"queue",
"rate_limits",
"reservations",
"result_adapter",
"result_ttl",
"retry",
"retry_classifier",
"singleton",
"start_to_close",
"unique_for",
"unique_states",
"wants_ctx",
)
on_retry_exhausted_timeout
instance-attribute
¶
fn
property
¶
Return the underlying handler.
May be sync (def) or async (async def). Direct
invocation through :meth:__call__ is the safer path because
it handles both shapes and enforces the declared signature.
__call__
async
¶
Direct invocation — bypasses enqueue, runs the handler in-process.
Pass a :class:JobContext only when the registered handler
declared one (when :attr:wants_ctx is True); calling with
a context against a no-ctx handler raises :class:TypeError,
and calling without a context against a ctx handler also
raises :class:TypeError.
deps mirrors the keyword arguments the worker's
dependency-injection pass supplies in production. Tests may
pass them explicitly; the call site is responsible for
matching the names recorded in :attr:dependencies. Missing
dependencies surface as a runtime TypeError from Python's
argument binding.
Production callers go through :meth:JobsClient.enqueue.
Source code in src/taskq/actor.py
ActorConfigRow
dataclass
¶
ActorConfigRow(
actor: str,
max_concurrent: int | None,
max_pending: int | None,
queue: str,
result_ttl: float | None,
metadata: dict[str, object],
updated_at: str,
)
Snapshot of one {schema}.actor_config row.
DeregisterResult
dataclass
¶
DeregisterResult(
actor: str,
queue: str,
actor_config_deleted: bool,
schedules_disabled: int,
jobs_cancelled: int,
terminal_jobs_remaining: int,
queue_purged: bool,
)
Outcome of a deregister_actor call.
actor_config_deleted is always True — if the row is not found,
deregister_actor raises :class:ActorNotFoundError instead of
returning a result with False. The field is retained for API
contract clarity and consumer assertions.
PgCredential
dataclass
¶
A Postgres credential issued by a rotating-credential provider.
password is always required (a token or dynamic password).
username, when set, overrides the DSN's userinfo user - needed by
providers that issue a fresh username alongside the password (e.g.
Vault dynamic DB creds). When None, the DSN's existing user is
preserved.
PgCredentialProvider ¶
Bases: Protocol
Provides rotating Postgres credentials on demand.
Implementations fetch a fresh token / dynamic username+password each
call. Called by :func:make_pg_pool_factory /
:func:make_dedicated_conn_factory once at pool / connection
construction (to resolve user= and fail fast), and then again for
every physical connection asyncpg opens thereafter - not on each
acquire(), which hands back an already-authenticated connection
from the pool. Implementations are expected to cache and only hit the
issuing service when the cached credential is near expiry.
RedisCredential
dataclass
¶
RedisCredentialProvider ¶
Bases: Protocol
Provides rotating Redis credentials on demand.
Implementations fetch a fresh (username, token/password) each call.
Called by :func:make_redis_client_factory on every reconnect via
the redis-py CredentialProvider adapter.
Backend ¶
Bases: Protocol
Contract that both PostgresBackend and InMemoryBackend satisfy.
46 async methods plus two sync methods (subscribe_wake and
subscribe_cancel_wake) (48 methods total) covering enqueue,
dispatch, heartbeat, terminal writes, attempt history, cancel
signals, scheduling / sweeps, read, NOTIFY hook, schedule CRUD,
and batch operations. Method order grouped for review-grep
ergonomics.
Why monomorphic (no Generic[P, R]): the backend is the DB
adapter boundary. Payloads are stored as dict[str, object] (the
JSONB payload column) regardless of the actor's typed payload
model. Generic parameters here would propagate P and R into
every method (dispatch_batch, mark_succeeded, etc.) with no
safety benefit at the storage layer. The worker consumer
reconstructs the typed JobContext[P] at dispatch time using
ActorRef.payload_type.model_validate(row.payload).
supports_transactional_simulation
class-attribute
¶
Whether this backend simulates transactional sub-enqueue via a buffer (True) or relies on real database transactions (False).
PostgresBackend returns False — its real PG transaction provides
the atomicity guarantee directly: sub-job INSERTs run on the open
LOOP-scope connection and are rolled back along with the parent's
writes if the actor raises.
InMemoryBackend returns True — it has no real transaction
concept, so SubJobEnqueuer buffers EnqueueArgs and flushes on
actor success / discards on failure. A third-party Backend
implementation that wants transactional simulation in tests can opt
in by overriding this to True.
enqueue
async
¶
Source code in src/taskq/backend/_protocol.py
enqueue_batch
async
¶
enqueue_batch(
args_list: list[EnqueueArgs],
*,
connection: Connection | None = None,
) -> list[JobRow]
Insert multiple jobs in a single batched operation.
All items in args_list must be validated before calling this method — the backend does not re-validate payloads. The list must be non-empty and contain at most 1000 items (enforced by the client layer).
Returns one :class:JobRow per item in args_list, in the same
order. For idempotency-key collisions the existing row is
returned; its id will differ from the requested args.id.
Source code in src/taskq/backend/_protocol.py
enqueue_batch_fast
async
¶
Insert multiple jobs via the COPY FROM protocol for maximum throughput.
COPY cannot evaluate expressions or handle conflicts, so the write
is two statements inside one transaction: a bare COPY of the
domain-insensitive columns, then a corrective UPDATE
(enqueue_batch_fast_fixup) that stamps/decides the
clock-sensitive ones — status, scheduled_at,
schedule_to_close, result_expires_at — from the database
clock (clock_timestamp()); created_at takes its DDL
default (now()). Nothing is observable half-fixed: both
statements commit or abort together.
Consequences of the COPY-no-conflicts shape:
scheduled_at=Nonemeans immediate — the fixup's server-side CASE stamps it and decidespending/scheduled(the same single-arbiter contract as :meth:enqueue/:meth:enqueue_batch).schedule_to_close_interval/result_ttlare anchored to the server clock at ENQUEUE time by the fixup — a future-scheduled item with a short interval can therefore fail DeadlineExceeded before it is ever dispatched.- A duplicate
idempotency_key— within the batch or already stored — violates the unique index and aborts the ENTIRE batch (all-or-nothing atomicity; nothing is written).
Returns the count of rows written. On success this is exactly
len(args_list) — this path never deduplicates, so the count
never includes pre-existing rows. The in-memory mirror implements
the same contract: duplicates raise
asyncpg.UniqueViolationError before any row is written, and
the count is the number of items.
This is a performance-focused variant of :meth:enqueue_batch
(which DOES deduplicate idempotency-key collisions via ON
CONFLICT). Use for bulk import / backfill with 10K+ rows where
collision handling is not needed. Max batch size is 50 000
(client-enforced).
Source code in src/taskq/backend/_protocol.py
enqueue_with_conn
async
¶
Enqueue a job using the supplied connection.
The connection MUST already be in an open transaction managed by
the caller — this method does NOT issue BEGIN/COMMIT. The
autonomous variant enqueue(args) acquires its own connection
and opens a transaction internally.
Source code in src/taskq/backend/_protocol.py
dispatch_batch
async
¶
heartbeat_jobs
async
¶
extend_reservation_leases
async
¶
mark_succeeded
async
¶
mark_succeeded(
job_id: JobId,
worker_id: UUID,
result: dict[str, object] | None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
fallback_result_ttl: timedelta | None = None,
) -> bool
Mark a job succeeded, computing result_expires_at at completion.
Expiry resolution, first match wins: a non-NULL stored
actor_config.result_ttl (operator-owned) applies; otherwise
fallback_result_ttl — the worker-side @actor(result_ttl=...)
literal, which the terminal-write SQL cannot see — applies;
otherwise the row's existing result_expires_at is kept. The
computed arms use clock_timestamp() — the wall-clock time the
write executes, not the transaction start — so neither a long
queue wait nor a long actor runtime can make a job complete
already expired and have its result reaped immediately.
Source code in src/taskq/backend/_protocol.py
mark_succeeded_with_conn
async
¶
mark_succeeded_with_conn(
conn: Connection,
job_id: JobId,
worker_id: UUID,
result: dict[str, object] | None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
fallback_result_ttl: timedelta | None = None,
) -> bool
Mark a job succeeded using the supplied connection.
Used by the consumer when a LOOP-scope asyncpg.Connection is
available so the success status update commits atomically with
the actor's writes and sub-job INSERTs in the same transaction.
The connection MUST already be in an open transaction; this
method does NOT open or close one. The autonomous variant
mark_succeeded(...) acquires its own connection.
fallback_result_ttl follows the same resolution rule as
:meth:mark_succeeded.
Source code in src/taskq/backend/_protocol.py
mark_failed_or_retry
async
¶
mark_failed_or_retry(
job_id: JobId,
worker_id: UUID,
error_info: ErrorInfo,
retry_delay: timedelta | None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
) -> JobRow
Mark a running job failed, or schedule a retry retry_delay later.
retry_delay=None is the terminal-fail arm (status='failed',
the original error_info persisted). A non-None delay is applied
by the backend's own clock, never the caller's: scheduled_at =
now() + delay and the scheduled/pending status derive from
the delay alone (zero → immediate). The same statement arbitrates
the schedule_to_close deadline server-side — when
clock_timestamp() + delay would land past the deadline, the row
is failed with error_class='DeadlineExceeded' instead of
retried — so app↔DB clock skew can neither void the retry backoff
nor kill a job whose deadline has not actually passed.
Source code in src/taskq/backend/_protocol.py
mark_cancelled
async
¶
write_cancel_escalation
async
¶
mark_abandoned
async
¶
mark_snoozed
async
¶
mark_snoozed(
job_id: JobId,
worker_id: UUID,
delay: timedelta,
*,
metadata_update: dict[str, object] | None = None,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
outcome: AttemptOutcome = "snoozed",
) -> Literal["scheduled", "failed", "noop"]
Source code in src/taskq/backend/_protocol.py
mark_retry_after
async
¶
mark_retry_after(
job_id: JobId,
worker_id: UUID,
delay: timedelta,
*,
consume_budget: bool = True,
progress_seq: int = 0,
progress_state: dict[str, object] | None = None,
) -> Literal[
"scheduled",
"failed:DeadlineExceeded",
"failed:MaxAttemptsExceeded",
"noop",
]
Source code in src/taskq/backend/_protocol.py
write_attempt
async
¶
Source code in src/taskq/backend/_protocol.py
get_attempts
async
¶
Source code in src/taskq/backend/_protocol.py
get_events
async
¶
Source code in src/taskq/backend/_protocol.py
poll_reclaim_events
async
¶
poll_reclaim_events(
after_id: int,
limit: int = DEFAULT_RECLAIM_POLL_LIMIT,
*,
visibility_delay: timedelta | None = None,
) -> list[EventRow]
Return up to limit crash-reclaim events with event_id >
after_id, ascending — the durable cursor behind
TaskQ.watch_reclaims.
An event can be silently missed if a job_events writer
transaction stays open longer than the visibility-delay margin
between its INSERT and its COMMIT — ids are allocated at INSERT
time but transactions commit out of order, so a late-committing
lower-id row can land behind an already-advanced cursor. Rows
are therefore held back by a trailing-watermark filter
(visibility_delay; backend-configured default when None —
see :data:taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY for
the exact assumption and its violation modes, and
PostgresBackend.check_reclaim_visibility_delay_risk for the
diagnostic that makes a violation operator-visible).
Source code in src/taskq/backend/_protocol.py
write_cancel_request
async
¶
cancel_where
async
¶
Cancel all jobs matching filter in a set-based operation.
Pending/scheduled jobs → terminal 'cancelled'. Running jobs → cancel_phase=1 (cooperative cancel + NOTIFY).
The filter's limit, cursor, and order_by fields are
ignored — this is a bulk write, not a paginated read.
Guardrail: the client layer (:meth:JobsClient.cancel_where)
rejects empty filters (no predicates) with
:class:EmptyFilterError. Backend implementations receive a
filter that has already been validated. A direct backend call
with JobFilter() renders WHERE TRUE and cancels the
entire table — callers using the backend directly are
responsible for validating the filter.
Returns a :class:BulkCancelResult with counts and affected IDs.
Source code in src/taskq/backend/_protocol.py
poll_cancel_flags
async
¶
retry_job
async
¶
Reset a terminal job (failed/crashed/cancelled) to pending.
Returns True if the job was retried, False if it was not
in a retryable state.
scheduled_to_pending
async
¶
Promote scheduled jobs whose scheduled_at has passed.
The backend's own clock is the arbiter (PG evaluates
scheduled_at <= clock_timestamp() server-side; InMemory
compares against its injected Clock). Returns the count of
promoted rows.
Source code in src/taskq/backend/_protocol.py
deadline_sweep
async
¶
Fail pending/scheduled jobs whose schedule_to_close has passed.
Transitions to failed with error_class='DeadlineExceeded',
arbitrated by the backend's own clock. Returns the count of swept
rows.
Source code in src/taskq/backend/_protocol.py
reclaim_expired_locks
async
¶
Reclaim running jobs whose lock has expired.
The expiry check is arbitrated by the backend's own clock; the grace parameters only widen the carve-out for jobs with an in-flight cancel request. Returns the count of reclaimed rows.
Source code in src/taskq/backend/_protocol.py
get
async
¶
Source code in src/taskq/backend/_protocol.py
list_jobs
async
¶
List jobs matching filters, returning at most filters.limit
rows in keyset-pagination order.
filters.status accepts a single :data:JobStatus or a
sequence of statuses; filters.active is a meta-filter for
non-terminal (True) or terminal (False) statuses —
'active' here means 'not yet finished', not Celery's 'currently
executing'. See :class:JobFilter for details.
Source code in src/taskq/backend/_protocol.py
count_pending_jobs
async
¶
Return pending+scheduled job counts per actor.
Returns a dict mapping actor name to count. Only actors with
at least one pending or scheduled job appear in the result.
Actors not in the result have a count of zero. The actors
list is used as an IN/ANY filter — pass all distinct actor
names from a batch to fetch all counts in one round-trip.
Source code in src/taskq/backend/_protocol.py
count_active_jobs
async
¶
Count non-terminal jobs (pending, scheduled, running) in the given queues.
Returns the total count across all specified queues. Used by the drain monitor to detect when queues are empty. An empty queues list returns 0.
Source code in src/taskq/backend/_protocol.py
get_actor_max_pending
async
¶
Return the stored actor_config.max_pending for every actor
with a row.
Key present with an int value: the stored (operator-owned)
limit. Key present with None: a row exists but the column is
NULL (a cleared override). Key absent: no stored row. Client-side
capacity resolution
(:class:taskq.client._capacity.ActorCapacityCache) treats
"absent" and "NULL" identically — both fall back to the
@actor(...) literal; the distinction is preserved here only
so observability callers can tell them apart.
This is the enqueue-path analog of the dispatch CTE's per-cycle
actor_config join: one small whole-table read, consumed
through a TTL-bounded cache so the hot path pays no per-enqueue
query.
Source code in src/taskq/backend/_protocol.py
subscribe_wake ¶
Source code in src/taskq/backend/_protocol.py
subscribe_cancel_wake ¶
Return an async context manager yielding a fresh asyncio.Event
that is set whenever a cancel NOTIFY arrives for any job.
The heartbeat loop uses this to interrupt its sleep immediately on cancel, rather than waiting for the next scheduled tick.
Source code in src/taskq/backend/_protocol.py
create_schedule
async
¶
Source code in src/taskq/backend/_protocol.py
list_schedules
async
¶
update_schedule
async
¶
delete_schedule
async
¶
Source code in src/taskq/backend/_protocol.py
enqueue_batch_atomic
async
¶
enqueue_batch_atomic(
items: Iterable[EnqueueArgs],
*,
batch_id: UUID,
queue: str,
batch_row: BatchRow | None,
finalizer_args: EnqueueArgs | None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> list[JobRow]
Source code in src/taskq/backend/_protocol.py
create_batch
async
¶
create_batch(
batch_id: UUID,
queue: str,
expected_size: int,
failure_threshold: int | None,
finalizer_job_id: UUID | None,
originating_actor: str | None,
*,
connection: Connection | None = None,
) -> None
Source code in src/taskq/backend/_protocol.py
increment_batch_failures
async
¶
reset_batch_failures
async
¶
abort_batch
async
¶
complete_batch
async
¶
get_batch
async
¶
list_batches
async
¶
count_batch_non_terminal
async
¶
BatchCounts
dataclass
¶
BatchCounts(
total: int,
pending: int,
succeeded: int,
failed: int,
cancelled: int,
crashed: int,
abandoned: int,
)
Live job-count aggregate for one batch.
Mirrors BatchCompletionStatus fields; defined at the protocol layer so backends do not import the client-side batch module.
BatchFilter
dataclass
¶
BatchFilter(
queue: str | None = None,
active: bool | None = None,
batch_id: UUID | None = None,
limit: int = 100,
cursor: str | None = None,
)
Filter parameters for Backend.list_batches.
Unlike JobFilter, this only carries fields relevant to batch queries: queue, active (status terminality), batch_id, limit, and cursor. Job-oriented fields (status, actor, tags, order_by, identity_key) are intentionally absent — using JobFilter for batch queries would silently ignore those fields, which is a type trap.
cursor is the same mechanism as :attr:JobFilter.cursor: an
opaque keyset token encoding the last row of the previous page, which
both backends must decode identically
(:func:~taskq.backend._cursor.encode_batch_cursor). It encodes
(created_at, id) where the job cursor encodes (priority,
scheduled_at, id) — one field fewer, same |-delimited shape.
created_at alone is not a total order (the column defaults to
now(), the transaction timestamp, so one
enqueue_batch_atomic stamps every row it writes identically), so
id is the tiebreaker; it is UUIDv7 and therefore time-ordered,
which lets both columns sort DESC together and makes the seam a
single row-wise comparison.
There is no order_by: list_batches has exactly one ordering,
so the cursor cannot disagree with it. list_jobs has three, and
binds each to its own cursor shape through
:class:~taskq.backend._cursor.JobOrdering for the same reason.
__post_init__ ¶
Source code in src/taskq/backend/_protocol.py
BatchRow
dataclass
¶
BatchRow(
id: UUID,
queue: str,
status: BatchStatus,
expected_size: int,
consecutive_failures: int,
failure_threshold: int | None,
finalizer_job_id: UUID | None,
originating_actor: str | None,
created_at: datetime,
completed_at: datetime | None,
metadata: dict[str, object],
)
Read-model of a taskq.batches row.
CancelPhase ¶
Bases: IntEnum
Phases of cooperative-then-forced cancellation.
Why IntEnum and not Literal[0, 1, 2]: the cancel-poll loop
performs arithmetic comparisons (db_phase >= 1,
active.cancel_phase < 2) that Literal[int] does not narrow
correctly under pyright strict. IntEnum subclasses int, so
every existing comparison continues to work, while the typed enum
carries the OTel attribute semantics (cancel_phase attribute on
transition counters) and prevents bare-int values like 99 from
slipping past the type checker.
Values NONE, COOPERATIVE, and FORCED are persistable —
they map directly to the PG cancel_phase column whose check
constraint is BETWEEN 0 AND 2. ABANDON_PENDING is an
in-process sentinel only: the cancel-poll loop sets it on
_ActiveJob to mark a job as queued for post-transaction
abandonment. It is never written to PG. Keeping it on the same
enum lets cancel_phase stay strongly typed end-to-end.
EventRow
dataclass
¶
EventRow(
event_id: int,
job_id: JobId,
occurred_at: datetime,
kind: Literal["state_change", "cancel_request"],
detail: dict[str, object],
)
Read-model of a taskq.job_events row.
Mirrors the job_events table shape: monotonic event_id,
the owning job, timestamp, event kind, and a detail payload.
JobFilter
dataclass
¶
JobFilter(
queue: str | None = None,
status: JobStatus | Sequence[JobStatus] | None = None,
actor: str | None = None,
identity_key: IdentityKey | None = None,
batch_id: UUID | None = None,
limit: int = 100,
cursor: str | None = None,
tags: tuple[str, ...] | None = None,
order_by: JobSortField | None = None,
active: bool | None = None,
)
Filter parameters for :meth:Backend.list_jobs and
:meth:Backend.cancel_where.
For cancel_where, the limit, cursor, and order_by
fields are ignored — a bulk cancel is not paginated. Use
:meth:has_predicates to check whether the filter has at least one
predicate before passing it to cancel_where.
Heads-up: active=True is not Celery's 'active' — Celery's
means 'currently executing' (running only), TaskQ's means 'not
yet finished' (pending + scheduled + running). Read the
active section below before relying on the name.
cursor is an opaque keyset-pagination token encoding the sort
columns of order_by's ordering from the last row of the previous
page -- (priority, scheduled_at, id) for the default,
(created_at, id) and (finished_at, id) for the others. Encode
it with :func:~taskq.backend._cursor.encode_job_cursor, passing the
same order_by: a cursor is only meaningful under the ordering that
produced it. Both backends must agree on cursor encoding and
comparison semantics.
batch_id is a :class:UUID. The PG backend converts it to its
canonical string form at the SQL boundary; the in-memory backend
compares the UUID directly. Keeping the typed shape here means
JobsClient.list(batch_id=UUID(...)) flows without an implicit
str(uuid) coercion.
status accepts either a single :data:JobStatus (backwards
compatible — e.g. JobFilter(status="pending")) or a sequence of
statuses (e.g. JobFilter(status=["pending", "running"])).
An empty sequence (status=[]) matches no jobs — it is not
treated as 'no filter'. Unknown status values raise
:class:ValueError in :meth:__post_init__, so untrusted input
fails identically on both backends instead of surfacing as a PG
enum-cast error or a silent empty result.
The PG backend renders a single status as status = $n and a
sequence as status = ANY($n); the in-memory backend performs a
membership check in both cases.
active is a meta-filter that selects statuses by terminality.
This is not Celery's 'active'. Celery/Flower use 'active' for
tasks currently executing on a worker (running only); here it
means 'not yet finished' — a superset that also includes work that
has not started yet:
active=True→ non-terminal statuses (pending, scheduled, running)active=False→ terminal statuses (succeeded, failed, cancelled, crashed, abandoned)active=None(default) → no status-terminality filter
The non-terminal set is derived from
:data:~taskq.backend.statemachine.ACTIVE_STATUSES, which is itself
derived from the state machine — adding a new non-terminal state
updates this filter automatically.
status and active are mutually exclusive; specifying both
raises :class:ValueError in :meth:__post_init__.
Usage examples::
JobsClient.list(JobFilter(status="pending"))
JobsClient.list(JobFilter(status=["pending", "running"]))
JobsClient.list(JobFilter(active=True))
__post_init__ ¶
Source code in src/taskq/backend/_protocol.py
has_predicates ¶
Return True if at least one filter predicate is set.
Used by JobsClient.cancel_where to reject empty filters that
would match the entire table. New predicate fields added to
JobFilter MUST be added here and in
build_filter_conditions — the two are kept in sync manually.
Non-predicate fields (limit, cursor, order_by) are
excluded by design.
Source code in src/taskq/backend/_protocol.py
JobPage
dataclass
¶
JobRow
dataclass
¶
JobRow(
id: JobId,
actor: str,
queue: str,
identity_key: IdentityKey | None,
fairness_key: str | None,
payload: dict[str, object],
payload_schema_ver: int,
status: JobStatus,
priority: int,
attempt: int,
max_attempts: int,
retry_kind: RetryKind,
schedule_to_close: datetime | None,
start_to_close: timedelta | None,
heartbeat_timeout: timedelta | None,
created_at: datetime,
scheduled_at: datetime,
started_at: datetime | None,
finished_at: datetime | None,
last_heartbeat_at: datetime | None,
locked_by_worker: UUID | None,
lock_expires_at: datetime | None,
cancel_requested_at: datetime | None,
cancel_phase: CancelPhase,
error_class: str | None,
error_message: str | None,
error_traceback: str | None,
progress_state: dict[str, object],
progress_seq: int,
result: dict[str, object] | None,
result_size_bytes: int | None,
result_expires_at: datetime | None,
idempotency_key: IdempotencyKey | None,
idempotency_scope: str,
trace_id: str | None,
span_id: str | None,
metadata: dict[str, object],
tags: tuple[str, ...],
)
Read-model of a taskq.jobs row. Every column the dispatch loop,
heartbeat, and terminal writes need appears as a typed field.
status uses a Literal union (8 values) matching the
job_status enum in 01.00.00_01_pre_initial.sql.
JobSortField ¶
Bases: Enum
Sort ordering for :meth:Backend.list_jobs via :attr:JobFilter.order_by.
SCHEDULED_AT_ASC (and the default None) preserve the canonical
dispatch-friendly ordering — priority DESC, scheduled_at ASC, id ASC —
so existing list_jobs callers see no behaviour change.
CREATED_AT_DESC and FINISHED_AT_DESC serve "latest run by business
key" queries: newest-created first and most-recently-finished first
(NULLS LAST) respectively.
Every ordering pages with a cursor. Each one orders id with its
primary column rather than against it -- created_at DESC, id DESC,
not created_at DESC, id ASC -- which is what makes the page seam a
single row-wise comparison. id is UUIDv7 and therefore
time-ordered, so running it with a timestamp column reorders nothing
in practice. The ordering and the cursor shape it pages with are one
object, :class:~taskq.backend._cursor.JobOrdering; a cursor encodes
the columns of the ordering it was produced under, so cursors are not
interchangeable between orderings.
ScheduleRecord ¶
Bases: BaseModel
Read-only snapshot of a cron schedule row from the database.
model_config = ConfigDict(frozen=True) enforces immutability per
public API discipline.
Clock ¶
Bases: Protocol
Time abstraction injected into Backends.
now() returns wall-clock UTC; monotonic() returns a
monotonically non-decreasing float suitable for local elapsed-time
deltas (e.g. cancel-phase tracking in).
SystemClock
dataclass
¶
Production clock delegating to the standard library.
now() calls datetime.now(UTC) (timezone-aware).
monotonic() calls time.monotonic().
BatchCompletionStatus ¶
Bases: BaseModel
Aggregated completion counts for a batch of jobs.
pending counts jobs still in flight (pending, scheduled,
or running status). is_complete is True when all jobs
have reached a terminal status.
BatchHandle ¶
Bases: BaseModel
Handle to a group of jobs inserted by a single
:meth:~taskq.client.JobsClient.enqueue_batch call.
Invariant: job_handles contains one
:class:~taskq.client.JobHandle per item in the original list
(including idempotency-key collisions that returned existing rows).
When a finalizer was enqueued, the finalizer handle is appended as
the last entry of job_handles AND set separately as
finalizer_handle. size is the number of non-finalizer items
(i.e. len(job_handles) - (1 if finalizer_handle is not None else 0)).
:meth:status queries the database for the current completion
counts of the batch.
job_handles
instance-attribute
¶
List of :class:~taskq.client.JobHandle instances, one per enqueued item.
finalizer_handle
class-attribute
instance-attribute
¶
The :class:~taskq.client.JobHandle for the finalizer job, or None
when no finalizer was enqueued. When set, the finalizer handle is also
appended as the last entry of :attr:job_handles for backward compat.
status
async
¶
Query live completion counts for all jobs in this batch.
Uses a JSONB containment query against the metadata column so
the jobs_metadata_gin_idx GIN index is used (@> is
supported by jsonb_path_ops). The query groups by status in a
single round-trip.
schema must match the schema used when the :class:PostgresBackend
was constructed (default "taskq").
Source code in src/taskq/batch.py
BatchSummary
dataclass
¶
BatchSummary(
batch_id: UUID,
queue: str,
status: BatchStatus,
expected_size: int,
consecutive_failures: int,
failure_threshold: int | None,
finalizer_job_id: UUID | None,
originating_actor: str | None,
created_at: datetime,
completed_at: datetime | None,
completion: BatchCompletionStatus,
)
One row from the batches table, augmented with live job counts.
EnqueueItem ¶
Bases: BaseModel
One item in a :meth:~taskq.client.JobsClient.enqueue_batch call.
actor_ref is an :class:~taskq.actor.ActorRef for any payload and
result type. payload is the Pydantic model that will be
serialized into the job row — it is validated by the actor's
payload_type inside :meth:~taskq.client.JobsClient.enqueue_batch
before any INSERT.
metadata is merged with the library-injected batch_id key
before the row is written; callers MUST NOT set metadata.batch_id
manually.
AbortBatchAfter
dataclass
¶
Bases: BatchFailurePolicy
Abort the batch after consecutive_failures consecutive failures.
should_abort(n) returns True when n >= consecutive_failures.
failure_threshold is set to consecutive_failures in
__post_init__ so the client can read it polymorphically via
policy.failure_threshold.
Running jobs are NOT cancelled by the abort — only pending and scheduled jobs are cancelled. Running jobs continue to completion. This matches the post-terminal-write hook design: the hook runs after the terminal write, so a job that was dispatched before the abort triggered will run to completion.
__post_init__ ¶
BatchFailurePolicy
dataclass
¶
Abstract base class for batch failure policies.
Subclasses implement :meth:should_abort to decide whether a batch
should be aborted given the current count of consecutive failures.
failure_threshold is the consecutive-failure count at which the
batch should be aborted. None disables abort (the default on the
base class). :class:AbortBatchAfter sets this to its
consecutive_failures value in __post_init__; custom policies
should set it to their computed threshold. The client reads
failure_policy.failure_threshold polymorphically — no
isinstance check is needed.
BulkCancelResult ¶
Bases: BaseModel
Structured outcome of a bulk cancellation request.
Returned by JobsClient.cancel_where() so callers can inspect
how many jobs were cancelled directly (pending/scheduled → terminal
'cancelled') vs how many had cooperative cancel requested (running →
cancel_phase=1).
cancelled_directly
instance-attribute
¶
Count of pending/scheduled jobs moved straight to terminal 'cancelled'.
cancel_requested
instance-attribute
¶
Count of running jobs with cancel_phase=1 set (cooperative cancel).
cancelled_ids
instance-attribute
¶
IDs of jobs cancelled directly (pending/scheduled → cancelled).
cancel_requested_ids
instance-attribute
¶
IDs of running jobs with cancel requested.
CancelResult ¶
Bases: BaseModel
Structured outcome of a cancellation request.
Returned by JobsClient.cancel() so callers can inspect whether
the cancellation was initiated and what the status transition was.
JobEvent ¶
Bases: BaseModel
A single event yielded by :meth:TaskQ.stream.
Represents a point-in-time snapshot of a job's observable state.
Yielded on every status transition or progress update; the final
event always has terminal=True.
The progress_state and progress_seq fields reflect the last
values written by the worker. They are None / 0 until the
worker emits a progress update.
Serialises cleanly to JSON via model_dump() for SSE or WebSocket
fanout — fields are deliberately flat so the caller can forward the
event without transformation::
async for event in tq.stream(job_id):
await websocket.send_json(event.model_dump())
JobHandle ¶
JobHandle(
*,
row: JobRow,
result_adapter: TypeAdapter[R],
was_existing: bool,
client: JobsClient | None = None,
backend: Backend | None = None,
_redis_client: Redis | None = None,
_settings: TaskQSettings | None = None,
)
Typed handle to a single enqueued job.
Created by :class:JobsClient methods (:meth:~JobsClient.enqueue,
:meth:~JobsClient.get) or by :class:SubJobEnqueuer (with
backend= only). The type parameter R flows from the
actor's declared return type through :class:ActorRef into this
handle: JobHandle[OrderResult] for an actor returning
OrderResult, JobHandle[None] for fire-and-forget actors.
At least one of client or backend must be supplied. When
client is provided, _backend is filled from
client.backend. When only backend is provided, the four
read-back methods (:meth:status, :meth:refresh,
:meth:attempts, :meth:cancel) raise :class:RuntimeError
because they require the client's higher-level coordination.
:meth:wait always works (it reads through _backend directly).
Why result_adapter: TypeAdapter[R] is a constructor arg: pyright
only infers R for a generic class when the type parameter
appears in at least one field or method signature. The adapter is
that field — without it R would be phantom and inference would
silently fall back to Unknown.
Source code in src/taskq/client/_handle.py
row
property
¶
The last :class:JobRow this handle observed.
Seeded at construction with the row the creating call fetched,
and advanced by every successful row fetch through the handle —
:meth:refresh, :meth:status, and :meth:wait's polling loop
each record the row they just read. :meth:progress_stream
does not advance it — its Redis path fetches no rows, and
advancing only on the PG fallback would make the semantics
backend-dependent. Reading the property costs no backend round
trip: handle = await tq.get(id) followed by handle.row
is the single-read pattern for full row state, and a
long-lived handle's row stays current as its owner
refreshes or waits.
The row is returned by reference (it is frozen and backends hand the handle an isolated row); repeated reads between fetches return the same object.
status
async
¶
Return the current status of this job (live read).
Cheap, non-blocking: a single backend.get and a status
projection. No polling. Use this when you want to know the
state without waiting for a terminal transition. Advances
:attr:row to the fetched row.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
this handle was constructed without a
:class: |
Source code in src/taskq/client/_handle.py
refresh
async
¶
Re-read the row from the backend and return the raw
:class:JobRow.
Useful for callers that want full row state (timestamps,
attempt counts, error metadata) without going through
:meth:wait. Does not block on terminal state — returns the
current row whatever its status. Advances :attr:row to the
fetched row, so after a refresh handle.row and the return
value are the same row.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
this handle was constructed without a
:class: |
Source code in src/taskq/client/_handle.py
attempts
async
¶
Return the attempt rows for this job, ordered by attempt number.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
this handle was constructed without a
:class: |
Source code in src/taskq/client/_handle.py
cancel
async
¶
Delegate to :meth:JobsClient.cancel.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
this handle was constructed without a
:class: |
Source code in src/taskq/client/_handle.py
wait
async
¶
Block until the job reaches a terminal status, then return R.
Returns the actor's return value, validated through
:attr:result_adapter. The result type is R exactly —
never R | None. Missing or failed results raise. Advances
:attr:row to each row the polling loop fetches — on return,
the terminal row the result was extracted from.
Raises:
| Type | Description |
|---|---|
ResultUnavailable
|
terminal state reached but no result was
stored (result TTL expired, actor returned |
JobFailed
|
the job ended in a non-success terminal state
( |
TimeoutError
|
|
Source code in src/taskq/client/_handle.py
progress_stream
async
¶
Stream live progress events for this job.
When Redis is configured, subscribes to the per-job Redis pub/sub
channel and yields :class:~taskq.progress.ProgressEvent objects in
real time. When Redis is not available, falls back to polling Postgres
at 500 ms intervals and synthesising events from row diffs.
Raises :class:NotImplementedError when the in-memory backend is
detected — the in-memory backend does not support pub/sub.
Does not advance :attr:row — the Redis path fetches no rows,
and advancing only on the PG fallback would make the semantics
backend-dependent.
Yields events until a terminal=True event is produced.
Source code in src/taskq/client/_handle.py
JobsClient ¶
JobsClient(
backend: Backend,
*,
clock: Clock | None = None,
settings: TaskQSettings | None = None,
capacity_cache_ttl: float = DEFAULT_CAPACITY_CACHE_TTL,
)
Public API for job operations.
Delegates to the injected :class:Backend and wraps results in
typed :class:JobHandle[R] instances. The client owns the
payload-serialization step that turns a typed P into the
dict[str, object] carried by :class:EnqueueArgs; the backend
sees only erased payloads.
Source code in src/taskq/client/_jobs.py
backend
property
¶
The underlying :class:Backend this client delegates to.
Exposed so :class:JobHandle can read the backend through the
client without accessing the private _backend attribute.
close
async
¶
invalidate_actor_capacity_cache ¶
Drop the cached actor_config.max_pending snapshot.
The next enqueue refreshes from the backend instead of waiting
out the TTL. Not needed in normal operation (staleness is
bounded by capacity_cache_ttl, default 5s); intended for
tests and for tooling that knows it just changed the table and
cannot wait out the TTL.
Source code in src/taskq/client/_jobs.py
enqueue
async
¶
enqueue(
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]
Enqueue a job for the given actor and return a typed handle.
The payload is serialized through ref.payload_type so the
EnqueueArgs.payload carried over the backend boundary is a
plain dict[str, object] ready for the JSONB column. The
returned :class:JobHandle[R] carries ref.result_adapter so
:meth:JobHandle.wait can validate the stored result back to
R.
The metadata.singleton key is reserved by the library for
singleton enforcement. When ref.singleton is True the
library unconditionally writes metadata.singleton = True,
overriding any caller-supplied value. Callers MUST NOT set
metadata.singleton manually.
max_pending:
-
When the actor's effective
max_pendingis set, a pre-flight count ofpending+scheduledjobs for the actor is compared to the limit. Ifcount >= max_pending, :class:MaxPendingExceededErroris raised synchronously — the caller decides whether to retry, fail, or wait; the library does not block on capacity. -
The effective limit is operator-owned: a non-NULL stored
actor_config.max_pending(set viataskq actor-config set --max-pending) wins over the@actor(max_pending=...)literal; a cleared or absent stored value falls back to the literal. The client reads the stored value through a TTL-bounded cache (default 5s staleness; see :class:taskq.client._capacity.ActorCapacityCache), so an operator change takes effect fleet-wide within seconds without any redeploy or restart. -
Evaluation order at enqueue:
unique_fordedup → singleton pre-flight →max_pendingcount check →idempotency_keyINSERT → job INSERT. Aunique_forhit bypasses all remaining checks; a singleton collision fires beforemax_pendingto give the caller the more specificSingletonCollisionError. -
idempotency_keydoes not bypassmax_pending— the idempotency ON CONFLICT fires at step 5, after the max_pending check at step 3. Re-enqueuing with a duplicateidempotency_keywhen the queue is full raisesMaxPendingExceededError, not the deduplicated handle. Onlyunique_for(step 1) bypasses max_pending.
idempotency_key:
-
idempotency_keyis unique within itsidempotency_scope(composite(idempotency_scope, idempotency_key)uniqueness). The default scope (idempotency_scope=Noneor"") preserves the prior global-until-prune behavior exactly, so existing callers see zero behavior change. Passing an explicit scope (e.g. a run/batch/epoch id) lets two enqueues with the same business key in different scopes both succeed, decoupling the dedupe horizon fromprune_retention_*. -
Key length is bounded at
idempotency_key_max_bytes(TASKQ_IDEMPOTENCY_KEY_MAX_BYTES, default 1024 UTF-8 bytes) — the bound is the composite unique index's btree entry size, not a round number. Empty and whitespace-only keys raise :class:ValueErrorat the client boundary before any backend call. The same bound applies toidempotency_scope; an empty scope ("") is valid and equivalent toNone(the default/global scope). -
No time-based (TTL) dedupe window.
idempotency_scopedecouples the dedupe horizon fromprune_retention_*by namespace, not by time — there is noidempotency_ttlor equivalent "dedupe for the next N seconds" parameter. A key within a given scope still dedupes until pruned, exactly like the pre-scope global behavior, just scoped to that namespace. This is a deliberate scope decision, not an oversight: a real sliding-window TTL cannot be expressed as a single static unique index the way scope can — every mature job queue that offers one (Oban, River) either gives up the atomicINSERT ... ON CONFLICTfor a check-then-insert lock (weaker concurrency guarantee) or buckets time into the key itself (coarser, edge-artifact-prone semantics). If your use case genuinely needs "dedupe for the next hour, not forever," encode the window into the scope yourself (e.g. a time-bucketed scope string) until/unless a TTL parameter ships as a separate feature. -
Rolling-deploy note: if this schema is mid-upgrade (the
01.00.03_01_pre_idempotency_scope.sqlmigration applied but01.00.03_01_post_idempotency_scope_drop_old_index.sqlnot yet applied), reusing the sameidempotency_keyunder two differentidempotency_scopevalues raises :class:~taskq.exceptions.ScopedIdempotencyMigrationPendingErrorrather than silently dedupe against the wrong scope's job. The trigger is a key existing under a different scope, in either direction — an unscoped call reusing a key first written under a non-default scope raises it too. Only brand-new keys and same-scope repeats are unaffected. See that exception's docstring and the migration file's header comment for the full rationale.
unique_for:
-
unique_fordeduplication is best-effort. Concurrent enqueues for the same(actor, identity_key)may both insert; the dispatch CTE'srunning_identitiesfilter ensures only one runs. -
When either dedup mechanism matches an existing job,
JobHandle.was_existingisTrue. This field replaces the need for callers to inspect the row'screated_atto detect a dedup return.
Source code in src/taskq/client/_jobs.py
227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 | |
enqueue_batch
async
¶
enqueue_batch(
items: list[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
failure_policy: BatchFailurePolicy | None = None,
finalizer: EnqueueItem | None = None,
) -> BatchHandle
Enqueue multiple jobs in a single batched INSERT and return a
:class:~taskq.batch.BatchHandle.
All items share a single batch_id UUID written into each
job's metadata.batch_id field (as a string). When
batch_id is not supplied it is auto-generated as a UUIDv7 via
:func:~taskq._ids.new_job_id.
failure_policy:
When failure_policy is set (e.g.
:class:~taskq.batch_policy.AbortBatchAfter), a batches row
is created with the policy's failure threshold. After each child
job reaches a terminal state, the
:func:~taskq.batch.apply_batch_terminal_outcome hook inspects
the outcome: succeeded resets the consecutive-failure
counter; failed increments it and aborts the batch if the
threshold is reached. Aborting cancels all pending and scheduled
child jobs and sets the batch row to aborted.
finalizer:
When finalizer is set, a finalizer job is enqueued alongside
the batch. The finalizer is NOT stamped with batch_id
metadata (deadlock prevention — if it were, wait_for_batch
would count it as a child and the finalizer would wait for
itself). The batch row's finalizer_job_id column records the
link, and wait_for_batch automatically excludes that job
from counts. The finalizer is dispatched immediately; the
in-actor wait_for_batch snooze pattern gates on child-job
completion.
Transactional enqueue:
When failure_policy or finalizer is set and
connection is None, the entire operation (batch row +
all child jobs + finalizer) is inserted in a single transaction
via :meth: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.
Validation rules:
len(items) == 0raises :class:ValueError.len(items) > MAX_BATCH_SIZEraises :class:ValueError.- ALL payloads are validated before any INSERT. A single failure
raises :class:
~taskq.exceptions.PayloadValidationErrorand leaves no rows inserted.
max_pending:
One aggregated SELECT actor, count(*) … WHERE actor = ANY($1)
GROUP BY actor is issued for the entire batch. Per-actor
effective limits (operator-owned stored value when set, else the
@actor(...) literal — same resolution as :meth:enqueue)
are checked before the INSERT; any violation raises
:class:~taskq.exceptions.MaxPendingExceededError.
idempotency_key collisions:
Items whose idempotency_key collides with an existing row
return the existing :class:~taskq.client.JobHandle (same
semantics as single-item :meth:enqueue).
Source code in src/taskq/client/_jobs.py
421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 | |
enqueue_batch_streaming
async
¶
enqueue_batch_streaming(
items: Iterable[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
failure_policy: BatchFailurePolicy | None = None,
finalizer: EnqueueItem | None = None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> BatchHandle
Enqueue jobs from a lazy iterable in chunks, returning a single
:class:~taskq.batch.BatchHandle.
Unlike :meth:enqueue_batch, this method accepts an
:class:~collections.abc.Iterable (including generators) and
inserts in chunks of chunk_size (1-MAX_BATCH_SIZE). All items share
the same batch_id. Payloads are validated on the fly as
each chunk is built.
When failure_policy or finalizer is set and
connection is None, the entire operation is delegated to
:meth:Backend.enqueue_batch_atomic for single-transaction
atomicity. Otherwise, chunks are inserted via
:meth:Backend.enqueue_batch on the caller-owned connection,
and the batch row + finalizer are created as the last
statements.
Failure-policy counting limitation (caller-connection path):
the batch row is created AFTER all chunk inserts — it must carry
the final expected_size and create_batch is INSERT, not
upsert — so a child job enqueued on that connection that
reaches a terminal state BEFORE the row exists is not counted
toward failure_policy: increment_batch_failures finds
no row and returns (0, None, 0). The atomic
(no-connection) path is unaffected — its single transaction
makes the batch row and the child jobs visible together.
max_pending: NOT enforced on this path — unlike
:meth:enqueue_batch, which runs one aggregated per-actor
check before the INSERT, neither the chunked
:meth:Backend.enqueue_batch inserts nor the atomic delegation
consult max_pending. The caller is responsible for ensuring
the stream will not exceed actor limits (the same bulk-import
semantics :meth:enqueue_batch_fast discloses).
Source code in src/taskq/client/_jobs.py
675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 | |
get_batch
async
¶
Fetch a single batch row by ID.
Delegates to :meth:Backend.get_batch. Returns None when the
batch does not exist.
Source code in src/taskq/client/_jobs.py
list_batches
async
¶
List batches matching filter, returning :class:BatchSummary objects.
Delegates to :meth:Backend.list_batches and maps each
(BatchRow, BatchCounts) pair to a :class:BatchSummary
with a :class:BatchCompletionStatus derived from the live counts.
Source code in src/taskq/client/_jobs.py
enqueue_batch_fast
async
¶
enqueue_batch_fast(
items: list[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
) -> int
Enqueue jobs via COPY FROM protocol for maximum throughput.
WARNING — bulk-import semantics, not general-purpose enqueue:
this method does NOT enforce max_pending, does NOT detect or
reject idempotency-key collisions (a duplicate key aborts the
whole batch instead of being treated as "already enqueued"), and
returns a bare row count, not per-job handles — there is no
way to await, cancel, or otherwise reference an individual job
from the return value. Use :meth:enqueue_batch unless you
specifically need COPY-level throughput for a one-shot bulk
import/backfill and have already accounted for these gaps.
Returns the count of inserted rows — no :class:~taskq.batch.BatchHandle,
no per-job :class:~taskq.client.JobHandle instances.
Validation rules:
len(items) == 0raises :class:ValueError.len(items) > 50_000raises :class:ValueError.- ALL payloads are validated before any INSERT — a single failure
raises :class:
~taskq.exceptions.PayloadValidationError.
Tradeoffs vs enqueue_batch:
- No idempotency-key collision handling. A duplicate key
aborts the entire batch with
asyncpg.UniqueViolationError. Callers must pre-deduplicate. One carve-out: during the01.00.03pre→post migration window, a key reused across different scopes raises :class:~taskq.exceptions.ScopedIdempotencyMigrationPendingErrorinstead, matching the other enqueue paths. - No max_pending check. The caller is responsible for ensuring the batch won't exceed actor limits.
- No JobHandle instances. Only the inserted row count is
returned. Use
batch_idto query rows post-insert. - All-or-nothing atomicity. No partial success — the entire COPY fails on any constraint violation.
Use for bulk import / backfill with 1K-50K rows where throughput matters more than idempotency guarantees.
Source code in src/taskq/client/_jobs.py
1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 | |
get
async
¶
Look up a job by id.
Returns None when the job does not exist; otherwise wraps
the row in a :class:JobHandle[R]. The caller may supply
result_adapter because lookups by id do not carry actor
identity — typical sources are
my_actor.result_adapter (when reuniting with an actor) or
TypeAdapter(type(None)) (when only row metadata is needed).
When result_adapter is None it defaults to
TypeAdapter(type(None)), which is suitable for status-only
lookups.
Source code in src/taskq/client/_jobs.py
get_row
async
¶
Look up a job by id and return the raw :class:JobRow.
Mirrors :meth:get's contract — one backend.get, None
when the job does not exist — without the handle machinery or
result adapter. For callers that never need a
:class:JobHandle, this is the direct form; for the fresh-read
case that does want a handle, prefer get plus the handle's
row property (still a single round trip).
Source code in src/taskq/client/_jobs.py
list
async
¶
List jobs matching filter, returning a :class:JobPage.
filter.status accepts a single :data:JobStatus or a
sequence of statuses (e.g. JobFilter(status=["pending",
"running"])).
filter.active is a meta-filter — not Celery's 'active':
active=True selects non-terminal statuses (pending,
scheduled, running — 'not yet finished', not 'currently
executing') and active=False selects terminal ones. See
:class:JobFilter for full semantics.
next_cursor is returned for every ordering, encoded from the
columns that ordering actually sorts by, and is only None on
the last page.
Source code in src/taskq/client/_jobs.py
cancel
async
¶
Request cancellation of a job and return a :class:CancelResult.
Reads the row first via :meth:Backend.get. If the job does not
exist, raises :class:KeyError — matching Python's stdlib
idiom for "asked for an entry by id; it isn't there".
Then calls :meth:Backend.write_cancel_request and reads the
row again to capture the new status. The previous_status
reflects the row at the first read, not atomically at
write-time (TOCTOU per ).
: increments taskq.cancellation.requested exactly once
per call, regardless of cancellation_initiated outcome.
Source code in src/taskq/client/_jobs.py
cancel_where
async
¶
cancel_where(
filter: JobFilter,
reason: str | None = None,
*,
allow_empty_filter: bool = False,
) -> BulkCancelResult
Cancel all jobs matching filter in a single set-based operation.
Pending/scheduled jobs are moved straight to terminal 'cancelled'
(no running actor to cooperate with). Running jobs get
cancel_phase=1 set (cooperative cancel) — the worker's
heartbeat-driven cancel controller observes the phase change and
sets the in-process cancel_event.
Guardrail: a filter with no predicates (no queue, status,
actor, identity_key, batch_id, tags, or active) is rejected with
:class:EmptyFilterError unless allow_empty_filter=True is
passed.
Filter fields used: queue, status, actor,
identity_key, batch_id, tags, active. The
limit, cursor, and order_by fields are ignored.
Returns a :class:BulkCancelResult with counts and affected IDs.
Source code in src/taskq/client/_jobs.py
create_schedule
async
¶
create_schedule(
actor: str | ActorRef[P, R],
cron_expr: str,
*,
timezone: str = "UTC",
dst_strategy: DstStrategy = "skip",
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
name: str = "",
identity_key: IdentityKey | None = None,
enabled: bool = True,
) -> ScheduleHandle
Create a cron schedule. Raises :class:ValueError if both
payload_factory and static_payload are provided, or if
cron_expr is invalid.
The (actor, name) UNIQUE constraint means each (actor, name)
pair may have at most one schedule; a second create_schedule for
the same pair raises asyncpg.UniqueViolationError (PG) or
:class:ValueError (in-memory). Pass distinct name values to run
several cron schedules per actor (e.g. a per-property sync).
When identity_key is set, the cron loop propagates it to cron-fired jobs so they dedup against on-demand jobs for the same business key.
Does NOT validate actor existence at creation time — any string actor name is accepted (validation is deferred to fire time).
The first next_fire_at is seeded from the clock that
arbitrates its due-check: on a Postgres-backed client the PG
server clock is read first (one-row SELECT clock_timestamp()
via the backend's pool, mirroring the worker bootstrap), so
app↔DB clock skew cannot shift the fire chain; on a pool-less
(in-memory) client the seed comes from the client's injected
Clock. This matters permanently: the cron loop's normal path
recomputes every subsequent fire from the STORED fire time
(only a miss beyond cron_catch_up_window re-anchors on the
server clock), so the seed — not any per-tick correction —
fixes the chain's phase for the schedule's life.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dst_strategy
|
DstStrategy
|
How to handle DST gaps and overlaps.
|
'skip'
|
Source code in src/taskq/client/_jobs.py
1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 | |
list_schedules
async
¶
List cron schedules, optionally filtered by actor or enabled status.
Source code in src/taskq/client/_jobs.py
update_schedule
async
¶
update_schedule(
schedule_id: UUID,
*,
cron_expr: str | None = None,
enabled: bool | None = None,
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
clear_payload_factory: bool = False,
) -> ScheduleRecord
Update a cron schedule. Setting enabled=True clears
last_fire_error and resets consecutive_failures to 0.
Raises :class:ValueError if both payload_factory and
static_payload are provided, or if cron_expr is invalid.
To explicitly clear payload_factory (set the column to NULL),
pass clear_payload_factory=True — None for payload_factory
means "don't change this field."
When cron_expr changes, the recomputed next_fire_at is
seeded from the same clock as create_schedule (the PG
server clock on Postgres-backed clients; the client's injected
Clock in-memory) — the stored chain keeps its server-anchored
phase.
Source code in src/taskq/client/_jobs.py
delete_schedule
async
¶
TaskQ ¶
TaskQ(
*,
dsn: str | None = None,
pool: Pool | None = None,
pool_factory: PoolFactory | None = None,
pg_provider: PgCredentialProvider | None = None,
schema: str = "taskq",
min_pool_size: int = 1,
max_pool_size: int = 5,
redis_url: str | None = None,
redis_client: Any | None = None,
pg_conn_factory: ConnFactory | None = None,
listen_conn: Connection | None = None,
poll_timeout: float = 30.0,
reclaim_event_visibility_delay: timedelta | None = None,
)
Postgres-backed TaskQ client.
Manages a connection pool and exposes job operations directly. Supports
both the async context manager pattern and explicit open() / close()
for frameworks like FastAPI that manage their own lifecycle.
Parameters¶
dsn:
Postgres DSN string. Mutually exclusive with pool.
pool:
An already-open asyncpg.Pool. The caller retains ownership;
close() will not close it.
pool_factory:
A zero-arg async factory returning an asyncpg.Pool - the same
:data:~taskq.connections.PoolFactory the worker takes via
:class:~taskq.connections.WorkerConnections. TaskQ invokes it at
open() and owns the result (close() closes it), and
:meth:reload_credentials re-invokes it to rotate the pool in place.
This is the client-side equivalent of a <role>_pool_factory:
pair it with :func:taskq.auth.make_pg_pool_factory for a
rotating-credential deployment. Mutually exclusive with dsn and
pool.
pg_provider:
A :class:~taskq.auth.PgCredentialProvider. Sugar for
pool_factory=make_pg_pool_factory(dsn, pg_provider,
min_size=min_pool_size, max_size=max_pool_size) - it is exactly
that call and nothing more, so the two paths share one mechanism.
Requires dsn; use pool_factory directly when you need the
factory's other hooks (init, server_settings,
command_timeout).
schema:
TaskQ schema name. Defaults to "taskq".
min_pool_size:
Minimum pool connections. Only used when dsn is provided.
max_pool_size:
Maximum pool connections. Only used when dsn is provided.
redis_url:
Redis URL string. Mutually exclusive with redis_client.
The library creates and owns the Redis client; close() will
close it.
redis_client:
An already-open redis.asyncio.Redis client. The caller retains
ownership; close() will not close it. Mutually exclusive with
redis_url.
pg_conn_factory:
A zero-arg async factory returning an asyncpg.Connection for the
LISTEN/NOTIFY transport used by :meth:stream. Mutually exclusive
with listen_conn. Takes precedence over dsn when set. Use
this when you have no DSN (e.g. AAD-managed-identity auth) but still
want streaming. TaskQ owns and closes the connection produced by
the factory per stream() call.
listen_conn:
A pre-constructed asyncpg.Connection for the LISTEN transport.
Caller-owned; TaskQ does not close it. Mutually exclusive with
pg_conn_factory. Takes precedence over dsn when set. Use
this to share a dedicated LISTEN conn across callers.
poll_timeout:
Maximum seconds to wait between transport wakeups before re-fetching
job state. Defaults to 30.0.
Source code in src/taskq/client/_taskq.py
218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 | |
actors
property
¶
Actor configuration client — list, get, set capacity, deregister.
Raises RuntimeError if called before open() or outside an
async with block.
open
async
¶
Open the connection pool and prepare the client.
Called automatically by __aenter__. Safe to call explicitly
for frameworks that manage lifecycle outside an async with block.
Raises :class:RuntimeError if already open.
Source code in src/taskq/client/_taskq.py
close
async
¶
Close the client and release the pool if owned.
Called automatically by __aexit__. Safe to call explicitly.
No-op if already closed.
Source code in src/taskq/client/_taskq.py
reload_credentials
async
¶
Rebuild the pool from pool_factory and swap it in, live.
The client-side counterpart of
:func:taskq.worker.deps.reload_credentials, and the supported
replacement for reaching into _pool / _client: it invokes the
factory (which fetches a fresh credential), atomically points every
subsystem that holds a pool - the backend deps behind
enqueue/get/list/cancel, the :attr:actors client, and
the stream() LISTEN fallback - at the new pool, then closes the old
one with the same bounded drain used at close().
Call this on a schedule shorter than your credential's lifetime (an
Entra access token is typically ~60 min), or on any signal your issuer
gives you. It is not needed for the ordinary token refresh that
:func:taskq.auth.make_pg_pool_factory already does per physical
connection; it is how you drop sessions opened under a revoked
credential, and the only way to pick up a changed username, which
asyncpg resolves once per pool.
Raises :class:RuntimeError if the client is not open, or if the pool
is caller-owned (pool=) - TaskQ must never close a pool it does not
own, so the caller rotates that one itself.
If the factory fails, the exception propagates and the current pool is left untouched and serving: a transient token-endpoint outage must not turn into a client outage.
Source code in src/taskq/client/_taskq.py
__aenter__
async
¶
__aexit__
async
¶
enqueue
async
¶
enqueue(
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]
Enqueue a job and return a typed handle.
schedule_to_close (absolute datetime) is deprecated — it crosses
clock domains (the app clock that produced it vs the database clock
that evaluates it). Declare retry.time_budget on the actor
instead; the interval form is anchored to the database clock.
Source code in src/taskq/client/_taskq.py
enqueue_batch
async
¶
enqueue_batch(
items: list[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
failure_policy: BatchFailurePolicy | None = None,
finalizer: EnqueueItem | None = None,
) -> BatchHandle
Enqueue multiple jobs in a single batched INSERT.
Delegates to :meth:JobsClient.enqueue_batch; see its docstring
for validation rules and idempotency-key collision semantics.
Source code in src/taskq/client/_taskq.py
enqueue_batch_streaming
async
¶
enqueue_batch_streaming(
items: Iterable[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
failure_policy: BatchFailurePolicy | None = None,
finalizer: EnqueueItem | None = None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> BatchHandle
Enqueue jobs from a lazy iterable in chunks.
Delegates to :meth:JobsClient.enqueue_batch_streaming; see its
docstring for chunk_size validation and streaming semantics.
Source code in src/taskq/client/_taskq.py
get_batch
async
¶
Fetch a single batch row by ID.
Delegates to :meth:JobsClient.get_batch. Returns None when
the batch does not exist.
Source code in src/taskq/client/_taskq.py
list_batches
async
¶
List batches matching filter, returning :class:BatchSummary objects.
Delegates to :meth:JobsClient.list_batches.
Source code in src/taskq/client/_taskq.py
enqueue_batch_fast
async
¶
enqueue_batch_fast(
items: list[EnqueueItem],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
) -> int
Enqueue jobs via COPY FROM protocol for maximum throughput.
Delegates to :meth:JobsClient.enqueue_batch_fast; see its
docstring for tradeoffs vs the regular :meth:enqueue_batch.
Source code in src/taskq/client/_taskq.py
get
async
¶
Look up a job by id. Returns None when the job does not exist.
Source code in src/taskq/client/_taskq.py
get_row
async
¶
Look up a job by id and return the raw JobRow — no handle.
Delegates to :meth:JobsClient.get_row: one backend read,
None when the job does not exist, mirroring :meth:get's
contract without the handle machinery.
Source code in src/taskq/client/_taskq.py
list
async
¶
List jobs matching filter, returning a :class:JobPage.
Delegates to :meth:JobsClient.list — note filter.active
is not Celery's 'active' ('currently executing'); it selects by
terminality ('not yet finished'). See :class:JobFilter.
Source code in src/taskq/client/_taskq.py
cancel
async
¶
Request cancellation of a job. Raises :class:KeyError if not found.
cancel_where
async
¶
cancel_where(
filter: JobFilter,
reason: str | None = None,
*,
allow_empty_filter: bool = False,
) -> BulkCancelResult
Cancel all jobs matching filter. See :meth:JobsClient.cancel_where.
Source code in src/taskq/client/_taskq.py
create_schedule
async
¶
create_schedule(
actor: str | ActorRef[P, R],
cron_expr: str,
*,
timezone: str = "UTC",
dst_strategy: DstStrategy = "skip",
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
name: str = "",
identity_key: IdentityKey | None = None,
enabled: bool = True,
) -> ScheduleHandle
Create a cron schedule. Delegates to :meth:JobsClient.create_schedule.
dst_strategy controls how DST gaps/overlaps are handled; see
:meth:JobsClient.create_schedule for the full semantics.
Source code in src/taskq/client/_taskq.py
list_schedules
async
¶
List cron schedules. Delegates to :meth:JobsClient.list_schedules.
Source code in src/taskq/client/_taskq.py
update_schedule
async
¶
update_schedule(
schedule_id: UUID,
*,
cron_expr: str | None = None,
enabled: bool | None = None,
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
clear_payload_factory: bool = False,
) -> ScheduleRecord
Update a cron schedule. Delegates to :meth:JobsClient.update_schedule.
Source code in src/taskq/client/_taskq.py
delete_schedule
async
¶
Delete a cron schedule. Delegates to :meth:JobsClient.delete_schedule.
stream
async
¶
Stream live state changes for a job as :class:JobEvent objects.
Yields one event per observable state transition (status change or
progress update), terminating automatically when the job reaches a
terminal state. The final event always has ``terminal=True``.
Usage::
async for event in tq.stream(job_id):
print(event.status, event.progress_state)
# loop exits automatically when event.terminal is True
# Or wire directly into a FastAPI SSE response:
async def event_generator():
async for event in tq.stream(job_id):
yield f"data: {event.model_dump_json()}
"
Raises
------
RuntimeError
Called before ``tq.open()`` or outside an ``async with`` block.
KeyError
The job does not exist.
RuntimeError
PG LISTEN transport requested but ``dsn`` was not provided at
construction (pool-only mode).
Source code in src/taskq/client/_taskq.py
watch_reclaims
async
¶
watch_reclaims(
after_id: int = 0, *, poll_timeout: float | None = None
) -> AsyncIterator[EventRow]
Stream fleet-wide crash-reclaim events as :class:EventRow objects.
Delivery guarantee. At-least-once, and gap-free only under a
bounded writer-transaction assumption: an event is silently and
permanently missed if a job_events writer transaction stays
open longer than reclaim_event_visibility_delay (default 2s —
see :data:taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY)
between its INSERT and its COMMIT. Ids are allocated at INSERT
time but transactions commit out of order, so a late-committing
lower-id row can land behind the cursor after the cursor has
already advanced past its position; no error is raised anywhere.
Sweep and terminal-write transactions are a handful of
single-round-trip statements, so 2s is a generous bound under
normal operation — but it is an assumption enforced by nothing in
the SQL, not a property the query guarantees. So this watcher
does not leave detection to chance: on a slow cadence
(_VISIBILITY_RISK_CHECK_INTERVAL, 60s) it runs the backend's
check_reclaim_visibility_delay_risk diagnostic (when the
backend implements it — PostgresBackend does) and logs a loud
watch-reclaims-visibility-delay-at-risk warning for every
long-open job_events writer it finds. That is a proxy
warning, not proof of an actual miss.
Yields job_events rows with kind='state_change' and
detail['reason']='lock_expired', ordered by the monotonic
event_id cursor ascending. to_state is 'pending' for
a retried reclaim, 'crashed' or 'cancelled' (cancel was
in-flight when the worker died) for a terminal one.
Cursor and duplicate semantics¶
The caller persists the last-seen event_id and passes it back
as after_id on resumption. Persist the cursor after
processing each event: a crash between processing and persisting
re-delivers that event on the next run — delivery is
at-least-once, so consumers must dedupe on event_id. The
cursor is a watermark, not a reference: pruning job_events
rows at or below it is always safe, but rows pruned before the
consumer reads them are gone for good — only prune rows older
than your slowest consumer's cursor. A cursor far behind after
a long outage drains at query speed (full batches are re-polled
immediately, not one batch per poll_timeout).
Shutdown and backpressure¶
This is a pull-based async generator: events are fetched only as
fast as the consumer iterates, so a slow consumer simply polls
slower — no internal buffer grows. To stop, break out of the
async for (or cancel the consuming task); generator cleanup
removes the LISTEN registration and closes any owned connection.
Transport¶
On Postgres with a LISTEN transport source (dsn,
pg_conn_factory, or listen_conn), the method LISTENs on
wake_channel(schema) purely as a low-latency wakeup, but
always polls backend.poll_reclaim_events(after_id) as the
durable source of truth — NOTIFY is an optimisation, never the
only path, and a dropped LISTEN connection degrades to (and
recovers from) polling automatically. Without a LISTEN
transport or on Redis-configured backends, a plain poll loop
against poll_reclaim_events runs on poll_timeout.
Usage::
cursor = await load_cursor() # your own durable store
async for evt in tq.watch_reclaims(after_id=cursor):
outstanding -= 1 # fan-out completion tracking
cursor = evt.event_id
await save_cursor(cursor) # persist AFTER processing
Raises¶
RuntimeError
Called before tq.open() or outside an async with block.
Source code in src/taskq/client/_taskq.py
777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 | |
ActorsClient ¶
Pool-wrapping facade for actor configuration operations.
Acquires a connection from the injected pool for each call, delegates
to taskq.actor_config_ops, and returns the result. The
caller must have opened the pool; this class does not manage its
lifecycle.
.. note::
This client is Postgres-only — it delegates to
:mod:taskq.actor_config_ops, which executes raw SQL against
the actor_config table. The :class:~taskq.backend._protocol.Backend
protocol does not include actor config operations, so
:class:~taskq.testing.InMemoryBackend does not support
ActorsClient.
Parameters¶
pool:
An open asyncpg.Pool. The caller retains ownership.
schema:
TaskQ schema name. Defaults to "taskq".
Source code in src/taskq/client/_actors.py
list
async
¶
List all stored actor_config rows, ordered by actor name.
get
async
¶
Get one actor_config row, or None if not found.
set_capacity
async
¶
set_capacity(
actor: str,
*,
max_concurrent: int | Unset | None = UNSET,
max_pending: int | Unset | None = UNSET,
result_ttl: float | Unset | None = UNSET,
) -> ActorConfigRow | None
Update capacity fields on an existing actor_config row.
Source code in src/taskq/client/_actors.py
deregister
async
¶
Deregister an actor with safety checks.
See :func:taskq.actor_config_ops.deregister_actor for
the full semantics.
Source code in src/taskq/client/_actors.py
SubJobEnqueuer ¶
SubJobEnqueuer(
loop_scope_resolved: Mapping[type, object] | None,
worker_pool: Pool | None,
backend: Backend,
*,
clock: Clock | None = None,
capacity_cache: ActorCapacityCache | None = None,
)
Enqueue sub-jobs from within an actor body.
Uses the LOOP-scope DB connection by default (transactional enqueue). Falls back to the worker pool if no LOOP-scope connection is registered. One instance per loop — survives across dispatches so the per-100-enqueue re-warning fires on the loop-level counter, not per-job.
Source code in src/taskq/client/_enqueuer.py
enqueue
async
¶
enqueue(
actor_ref: ActorRef[P, R],
payload: P,
*,
connection: 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,
_batch_id: str | 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]
Enqueue a sub-job. max_pending is a per-call limit resolved
against the operator-owned stored cap and the @actor(...)
literal: against a non-NULL stored actor_config.max_pending
the tighter of the two wins (min(stored, per_call) — an
explicit caller shedding load is never widened by an operator
override, and no code path can raise an operator's fleet cap);
with no stored value this parameter wins outright over the
literal (historical behavior — actor code may loosen its own
declaration).
_batch_id is a library-internal parameter used by
:meth:enqueue_batch to stamp batch_id into metadata after
:func:build_enqueue_args has stripped any caller-supplied
batch_id (H5 security boundary). Callers MUST NOT pass it.
Source code in src/taskq/client/_enqueuer.py
121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 | |
enqueue_batch
async
¶
enqueue_batch(
items: Sequence[EnqueueItem[Any, Any]],
*,
batch_id: UUID | None = None,
connection: Connection | None = None,
) -> list[JobHandle[Any]]
Enqueue a batch of sub-jobs sharing a single batch_id.
All items share a single batch_id UUID written into each
job's metadata.batch_id field (as a string). When batch_id
is not supplied it is auto-generated as a UUIDv7 via
:func:~taskq._ids.new_job_id — mirrors
:meth:~taskq.client.JobsClient.enqueue_batch. Pass an explicit
batch_id to correlate this batch with a caller-constructed
identifier (e.g. a finalizer job enqueued separately that needs to
reference the same batch).
Raises ValueError when items is empty or exceeds
MAX_BATCH_SIZE — the same guardrails
:meth:~taskq.client.JobsClient.enqueue_batch applies to the
identical operation one layer up. Without the empty check the
no-connection fallback loop would iterate zero items and return
[] silently. The backend binds every item
as 21 parallel array parameters to a single unnest INSERT in one
transaction, so an uncapped batch enqueued from inside a job body is
unbounded fan-out that bypasses the client-side guardrail.
Source code in src/taskq/client/_enqueuer.py
292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 | |
flush_buffer
async
¶
Flush buffered EnqueueArgs to the backend (in-memory simulation).
Called by the consumer on actor success, AFTER the LOOP-scope
transaction has committed. Per-item flush failures are collected
and re-raised as :class:~taskq.exceptions.SubEnqueueError after
the loop completes so callers can detect lost sub-jobs.
Source code in src/taskq/client/_enqueuer.py
discard_buffer ¶
drain_for_re_enqueue ¶
Return and clear both loop-scope and pending buffers for re-enqueue.
Source code in src/taskq/client/_enqueuer.py
WorkerConnections
dataclass
¶
WorkerConnections(
dispatcher_pool: Pool | None = None,
dispatcher_pool_factory: PoolFactory | None = None,
heartbeat_pool: Pool | None = None,
heartbeat_pool_factory: PoolFactory | None = None,
worker_pool: Pool | None = None,
worker_pool_factory: PoolFactory | None = None,
notify_conn: Connection | None = None,
notify_conn_factory: ConnFactory | None = None,
leader_conn: Connection | None = None,
leader_conn_factory: ConnFactory | None = None,
redis_client: Redis | None = None,
redis_client_factory: RedisFactory | None = None,
)
Per-role connection overrides for the worker.
Each role has a <role> (pre-constructed, caller-owned) and a
<role>_factory (zero-arg async factory, TaskQ-owned) slot.
Leave both None for DSN-based construction (the default).
Example — AAD-managed-identity worker::
from azure.identity.aio import DefaultAzureCredential
from taskq.aad import EntraIdProvider
from taskq.auth import make_pg_pool_factory
from taskq.connections import WorkerConnections
cred = DefaultAzureCredential()
provider = EntraIdProvider(cred)
connections = WorkerConnections(
dispatcher_pool_factory=make_pg_pool_factory(
settings.pg_dsn_direct, provider, max_size=settings.dispatcher_pool_size,
),
heartbeat_pool_factory=make_pg_pool_factory(
settings.pg_dsn_direct, provider,
max_size=settings.heartbeat_pool_size,
command_timeout=settings.heartbeat_command_timeout,
),
worker_pool_factory=make_pg_pool_factory(
settings.pg_dsn_pooled, provider, max_size=settings.worker_pool_size,
),
)
Example — share an app-wide pool (caller-owned)::
connections = WorkerConnections(worker_pool=app_state.pg_pool)
dispatcher_pool
class-attribute
instance-attribute
¶
Dispatcher pool (pg_dsn_direct role). Caller-owned if set.
dispatcher_pool_factory
class-attribute
instance-attribute
¶
Factory for the dispatcher pool. TaskQ-owned.
heartbeat_pool
class-attribute
instance-attribute
¶
Heartbeat pool (pg_dsn_direct, heartbeat_command_timeout). Caller-owned.
heartbeat_pool_factory
class-attribute
instance-attribute
¶
Factory for the heartbeat pool. TaskQ-owned. command_timeout is
your responsibility when overriding — set it on create_pool.
worker_pool
class-attribute
instance-attribute
¶
Worker pool (pg_dsn_pooled role). Caller-owned.
worker_pool_factory
class-attribute
instance-attribute
¶
Factory for the worker pool. TaskQ-owned.
notify_conn
class-attribute
instance-attribute
¶
Dedicated LISTEN connection. Caller-owned. TaskQ still issues LISTEN.
notify_conn_factory
class-attribute
instance-attribute
¶
Factory for the LISTEN connection. TaskQ-owned.
leader_conn
class-attribute
instance-attribute
¶
Dedicated advisory-lock connection. Caller-owned.
leader_conn_factory
class-attribute
instance-attribute
¶
Factory for the advisory-lock connection. TaskQ-owned.
redis_client
class-attribute
instance-attribute
¶
Redis client for progress fanout / rate limiting. Caller-owned.
redis_client_factory
class-attribute
instance-attribute
¶
Factory for the Redis client. TaskQ-owned.
__post_init__ ¶
Reject concrete + factory for the same role (configuration error).
Source code in src/taskq/connections.py
has_any ¶
True if any override (concrete or factory) is set.
Source code in src/taskq/connections.py
JobContext
dataclass
¶
JobContext(
job_id: UUID,
actor: str,
queue: str,
attempt: int,
worker_id: UUID,
payload: P,
jobs: SubJobEnqueuer,
log: BoundLogger,
span: Span | None = None,
cancel_event: Event = asyncio.Event(),
_abort_requested: Event = threading.Event(),
_progress_buffers: dict[UUID, _ProgressBuffer]
| None = None,
_redis_client: Redis | None = None,
_worker_settings: WorkerSettings | None = None,
_pending_publish_tasks: set[Task[None]] | None = None,
)
Per-job execution context handed to worker actors.
The cancel_event field is a plain :class:asyncio.Event — never
wrapped in a cancel scope or :class:asyncio.TaskGroup (
PEP 789 mitigation). The consumer constructs a fresh event per
attempt; the cancel-poll hook sets it on phase 1; user actor code
polls :attr:cancellation_requested or awaits cancel_event.wait().
payload is typed as the actor's payload model P. The worker
consumer validates the raw dict[str, object] payload from the
JobRow against actor_ref.payload_type before constructing the
context, so handlers see a fully-validated Pydantic instance.
jobs provides :class:SubJobEnqueuer for enqueuing sub-jobs
from within the actor body. The enqueuer resolves the database
connection via LOOP-scope DI → worker-pool fallback.
cancel_event
class-attribute
instance-attribute
¶
check_cancelled ¶
should_abort ¶
Synchronous cancellation check for sync actors (thread-safe).
Sync actors cannot await the async :attr:cancel_event, so
they poll this method cooperatively. The cancel controller sets
the underlying :class:threading.Event during phase 1.
Returns:
| Type | Description |
|---|---|
bool
|
|
bool
|
actor should return or raise immediately. |
Source code in src/taskq/context.py
progress
async
¶
progress(
*,
step: int | None = None,
percent: float | None = None,
detail: str | None = None,
data: dict[str, object] | None = None,
) -> None
Report incremental progress for this job.
Updates the in-memory coalesce buffer synchronously, then schedules a
best-effort kind="progress" Redis publish as a background task
when a client is connected — this call never blocks on the network.
Raises :class:~taskq.exceptions.ProgressTooLarge if the serialised
data payload exceeds WorkerSettings.progress_data_max_bytes.
All arguments are optional and merged last-writer-wins into the
accumulated pending_state. Intermediate calls between periodic
flush ticks are coalesced: only the latest value for each field
reaches Postgres. seq is strictly monotone across calls.
The Redis publish is genuinely fire-and-forget: it may complete out
of order relative to other in-flight publishes for the same job.
Consumers reading the SSE/pub-sub stream already discard any event
whose seq is not greater than the last one seen (see
:mod:taskq.web.progress), so out-of-order or dropped publishes
never corrupt displayed state — the buffer mutation above (and the
eventual Postgres flush) is the durable source of truth. Failures
publishing to Redis are logged and recorded as a metric, never
raised here.
Source code in src/taskq/context.py
93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | |
CronScheduleSpec
dataclass
¶
CronScheduleSpec(
actor: str,
cron_expr: str,
timezone: str = "UTC",
dst_strategy: DstStrategy = "skip",
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
name: str = "",
identity_key: IdentityKey | None = None,
enabled: bool = True,
)
Immutable specification for a cron schedule row.
Created by the :func:cron decorator or constructed directly for
register_cron(). payload_factory and static_payload are
mutually exclusive — setting both raises :class:ValueError at
construction time (via :func:cron).
dst_strategy controls how DST gaps and overlaps are handled.
See :data:DstStrategy for the semantics of each strategy.
ScheduleHandle
dataclass
¶
ScheduleHandle(
schedule_id: UUID,
actor: str,
cron_expr: str,
timezone: str,
enabled: bool,
next_fire_at: datetime,
_backend: Backend = field(),
dst_strategy: DstStrategy = "skip",
name: str = "",
identity_key: IdentityKey | None = None,
)
Immutable handle for a cron schedule, returned by JobsClient methods.
The handle fields are a point-in-time snapshot of schedule state.
Async methods delegate to the Backend injected at construction time
(not part of the public __init__ signature) via ScheduleUpdateArgs.
enable() passes ScheduleUpdateArgs(enabled=True); the backend
resets consecutive_failures=0 and last_fire_error=NULL when
enabled=True is set.
ActorConfigDriftError ¶
ActorConfigDriftError(
actor: str,
field: Literal["queue", "metadata"],
registered: str | dict[str, object] | None,
stored: str | dict[str, object] | None,
)
Bases: TaskQError
One actor whose registered structural config differs from the stored row.
Only queue and metadata are structural — a mismatch there means
a stale worker is routing an actor to the wrong place, which is a
correctness bug. Capacity fields (max_concurrent, max_pending,
result_ttl) are operator-owned and never raise this error; see
:func:taskq.worker.startup.sync_actor_config.
Source code in src/taskq/exceptions.py
ActorConfigDriftList ¶
Bases: TaskQError
Collected wrapper raised at worker startup when one or more actors have drift.
Source code in src/taskq/exceptions.py
ActorDeregistrationError ¶
ActorHasActiveJobsError ¶
ActorHasActiveJobsError(
actor: str,
active_count: int,
status_counts: dict[str, int],
*,
force: bool = False,
)
Bases: ActorDeregistrationError
Non-terminal jobs reference the actor.
Carries the count and per-status breakdown of the blocking jobs so the
caller can decide whether to cancel them first or use force=True.
When force is True, the message reflects that running jobs
cannot be cancelled by force=True — the caller must wait for
them to finish or cancel them individually first.
Source code in src/taskq/exceptions.py
ActorHasEnabledSchedulesError ¶
Bases: ActorDeregistrationError
Enabled cron schedules reference the actor.
Carries the schedule IDs so the caller can disable or delete them first.
Source code in src/taskq/exceptions.py
ActorNotFoundError ¶
Bases: ActorDeregistrationError
The actor_config row does not exist — nothing to deregister.
Currently raised only by :func:deregister_actor. Other ops
(get, set_capacity) return None for missing rows.
Source code in src/taskq/exceptions.py
BackpressureError ¶
Bases: TaskQError
Base class for synchronous enqueue-time backpressure signals.
Subclassed by SingletonCollisionError (singleton collision) and used directly by max_pending enforcement. The caller decides whether to retry, fail, or wait; the library does not block on capacity.
Source code in src/taskq/exceptions.py
BatchAbortedError ¶
Bases: TaskQError
A batch was aborted because consecutive failures exceeded the threshold.
Running jobs are NOT cancelled by the abort — only pending and scheduled jobs are cancelled. Running jobs continue to completion. This matches the post-terminal-write hook design: the hook runs after the terminal write, so a job that was dispatched before the abort triggered will run to completion.
Source code in src/taskq/exceptions.py
DependencyCycle ¶
Bases: TaskQError
A cycle was detected in the provider graph.
Source code in src/taskq/exceptions.py
DIError ¶
Bases: TaskQError
Base for DI engine errors not covered by startup-validation.
Raised by the solver at resolution time for malformed annotations (e.g. multiple Scope markers in one Annotated parameter) or unresolvable forward references in actor signatures. Distinct from MissingProvider / ScopeViolation / DependencyCycle, which are raised at startup validation.
EmptyBatchError ¶
Bases: TaskQError
A batch has fewer jobs than the expected minimum.
This can happen when jobs were pruned before wait_for_batch ran,
or when expected_size was set but jobs were never created. Pass
on_empty="ok" to wait_for_batch to suppress the no-batch-row
variant of this error.
Source code in src/taskq/exceptions.py
EmptyFilterError ¶
Bases: TaskQError
Raised when cancel_where is called with a filter that has no predicates.
A filter with no queue, status, actor, identity_key, batch_id, tags, or
active predicate would match every job in the table — almost certainly
a bug. The guardrail is intentionally loud: the caller must add at least
one predicate or explicitly bypass with allow_empty_filter=True.
Source code in src/taskq/exceptions.py
IllegalStateTransition ¶
Bases: TaskQError
Attempted to transition a job to a status not reachable from its current status.
Best-effort fast-path check only; the SQL WHERE clause is the authoritative serialization gate for concurrent writes.
Source code in src/taskq/exceptions.py
JobFailed ¶
Bases: TaskQError
:meth:JobHandle.wait saw a non-success terminal state.
Carries the row so callers can inspect status, error_class,
error_message, and error_traceback. Distinct from
:class:ResultUnavailable (which means terminal but no result
stored) and from the original actor exception (which is recorded on
the row, not raised).
Source code in src/taskq/exceptions.py
MaxPendingExceededError ¶
Bases: BackpressureError
Raised when an actor's max_pending queue-depth limit is reached.
current_count is the count of pending+scheduled jobs at the time
of the pre-flight check. max_pending is the configured limit.
The caller decides whether to retry, fail, or wait; the library does
not block on capacity.
Source code in src/taskq/exceptions.py
MissingProvider ¶
Bases: TaskQError
A type was injected but no provider is registered.
Source code in src/taskq/exceptions.py
PartialBatchError ¶
PartialBatchError(
*,
succeeded_count: int,
failed_items: list[tuple[int, Exception]],
total: int,
)
Bases: TaskQError
Raised when an autonomous enqueue_batch partially fails.
Items enqueued before the first failure are committed; remaining
items are not inserted. succeeded_count is the number of items
that were successfully enqueued. failed_items maps the index
of each failed item to its exception. total is the original
batch size.
Source code in src/taskq/exceptions.py
PayloadValidationError ¶
PayloadValidationError(
detail: str,
*,
actor: str | None = None,
payload_schema_ver: str | None = None,
validation_errors: list[dict[str, object]]
| None = None,
)
Bases: TaskQError
Pydantic validation failed at enqueue or dispatch.
At enqueue: raised before the row is inserted ('fail at the door'). At dispatch: causes the job to transition to 'failed' with error_class='PayloadValidationError'. Non-retryable in both cases regardless of the actor's retry policy.
Source code in src/taskq/exceptions.py
ProgressTooLarge ¶
Bases: TaskQError
Raised when progress data payload exceeds the configured size limit.
limit is the configured cap in bytes (WorkerSettings.progress_data_max_bytes).
actual is the serialised byte length of the data dict that was rejected.
Non-retryable: the caller must reduce the payload before retrying.
Source code in src/taskq/exceptions.py
ReservationUnavailable ¶
ReservationUnavailable(
bucket_name: str,
retry_after: timedelta,
*,
source: Literal[
"reservation", "rate_limit"
] = "reservation",
)
Bases: TaskQError
A ConcurrencyReservation slot could not be acquired.
When the upstream RateLimitDecision.retry_after is None, callers
MUST substitute DEFAULT_RESERVATION_BACKOFF. When it is
timedelta(0) (allowed decisions) callers MUST pass it through
unchanged — do NOT use a truthiness coalesce
(x or DEFAULT_RESERVATION_BACKOFF) because timedelta(0) is falsy
and would be wrongly replaced.
Source code in src/taskq/exceptions.py
ResultTooLarge ¶
Bases: TaskQError
Terminal result exceeded WorkerSettings.result_max_bytes.
Non-retryable: the actor already ran to completion and a re-run returns
the same oversized value, so retrying only burns the remaining attempts
(re-running the actor's side effects each time) before the job fails
anyway. Classified alongside PayloadValidationError in
:meth:taskq.retry.RetryClassifier.classify.
ResultUnavailable ¶
Bases: TaskQError
:meth:JobHandle.wait saw a terminal state but no usable result.
Causes:
- result TTL expired before the call;
- actor returned None while R is non-None
(treated as schema mismatch, not a value);
- row stored result=NULL for a non-success status.
Carries the row for inspection.
Source code in src/taskq/exceptions.py
RetryAfter ¶
Bases: TaskQError
Schedule retry at specific delay. Consumes retry budget by default.
Source code in src/taskq/exceptions.py
SchemaNotMigratedError ¶
Bases: TaskQError
Backend raised UndefinedTableError — the TaskQ schema is missing.
Translated by the client layer (:mod:taskq.client._jobs) from an
asyncpg.exceptions.UndefinedTableError on the enqueue/get/list/cancel
paths, so operators see an actionable message instead of a raw asyncpg
traceback. The original exception is chained via __cause__.
Source code in src/taskq/exceptions.py
ScopedIdempotencyMigrationPendingError ¶
ScopedIdempotencyMigrationPendingError(
*,
actor: str | None = None,
idempotency_key: str | None = None,
idempotency_scope: str | None = None,
detail: str | None = None,
)
Bases: TaskQError
idempotency_scope was used, but the schema has not yet had
01.00.03_01_post_idempotency_scope_drop_old_index.sql applied.
Between applying 01.00.03_01_pre_idempotency_scope.sql and its
post counterpart (the rolling-deploy window every worker's schema
passes through), BOTH the old global jobs_idempotency_key_uniq
index (on idempotency_key alone) and the new composite
jobs_idempotency_scope_key_uniq index (on (idempotency_scope,
idempotency_key)) exist simultaneously — this is deliberate, see the
"PHASE OBLIGATIONS" comment in the pre migration file, and is what
keeps pre-this-release code's unscoped ON CONFLICT (idempotency_key)
working unmodified during the window.
The cost of that safety: enqueuing the same idempotency_key under
two different idempotency_scope values satisfies the new
composite index's ON CONFLICT target (no conflict there — the
(scope, key) pair is new) but still violates the still-present old
global index, which is not covered by that ON CONFLICT target.
PostgreSQL raises UniqueViolationError for a conflict against a
non-arbiter unique index unconditionally — the library deliberately
does NOT catch that and silently fall back to a different scope's row,
because doing so would return the wrong job for the scope the caller
actually asked for, silently, which is a worse failure mode than a
loud, explicit error for a purely transitional migration-window
condition. Raised instead of letting the raw
asyncpg.UniqueViolationError propagate.
Any call — scoped or unscoped — is affected whenever its
idempotency_key already exists under a different scope: an
unscoped call that reuses a key first written under a non-default
scope raises this error just as a scoped call reusing an unscoped
key does (verified against live PostgreSQL). Only brand-new keys and
same-scope repeats are unaffected — a repeated key under the same
scope (including two unscoped calls, which share the default ''
scope) conflicts identically against both indexes for the exact same
row, which ON CONFLICT DO NOTHING on the composite index resolves
cleanly.
Resolution: confirm every worker is running the release that shipped
idempotency_scope, then apply
taskq migrate up --phase post (or a plain taskq migrate up) to
drop the old index and activate scoped dedupe — or avoid passing
idempotency_scope until that migration has run.
Source code in src/taskq/exceptions.py
ScopeViolation ¶
Bases: TaskQError
A provider depends on a shorter-lived scope than its own.
Source code in src/taskq/exceptions.py
SingletonCollisionError ¶
SingletonCollisionError(
actor: str,
blocking_job_id: UUID | None = None,
retry_after: timedelta | None = None,
)
Bases: BackpressureError
Raised when a singleton actor already has a job in pending/scheduled/running.
blocking_job_id is the UUID of the existing job from the Layer 1
pre-flight query; it is None when raised from the Layer 2
UniqueViolationError catch (the race path) because no pre-flight row
was fetched.
retry_after is computed from the blocking job's schedule_to_close
when available. It is None when the blocking job has no
schedule_to_close set, or when raised from the Layer 2 catch path.
The heartbeat_interval * 4 fallback is intentionally NOT
implemented — retry_after is computed from schedule_to_close only.
Callers who need a poll cadence when retry_after is None should poll on
their own schedule (research.md Gap 1, resolution path (a)). Reason:
heartbeat_interval is not available at the backend enqueue boundary;
propagating it would require enlarging the backend constructor surface and
is out of scope.
Source code in src/taskq/exceptions.py
Snooze ¶
Bases: TaskQError
Job returns control with new scheduled_at; does not consume retry budget.
Source code in src/taskq/exceptions.py
SubEnqueueError ¶
Bases: TaskQError
Raised by flush_buffer() when one or more buffered sub-job enqueues fail after parent commit.
failed_items carries each failed EnqueueArgs and the exception
that caused the enqueue to fail. The parent job has already been
marked succeeded — this exception signals that child jobs were lost.
Source code in src/taskq/exceptions.py
TaskQError ¶
Bases: Exception
Base for all library-raised exceptions.
WorkerOwnershipMismatch ¶
Bases: TaskQError
Terminal write predicate failed: job exists but is owned by a different worker.
Source code in src/taskq/exceptions.py
ErrorReporter ¶
Bases: Protocol
Vendor-neutral hook for routing terminal job failures to external systems.
Implementations capture the error and job row, then forward to a
vendor-specific backend (Sentry, Datadog, a DLQ, etc.). The library
calls :meth:report when a job reaches a terminal failure state —
either because retries were exhausted or because the error was
non-retryable.
The call is wrapped in a try/except by :func:invoke_error_reporter;
a failing reporter never crashes the worker. Reporter failures are
counted on the taskq.error_reporter.failures counter with a
reporter_type attribute.
Register an :class:ErrorReporter instance as a DI provider
(registry.register_value(ErrorReporter, Scope.PROCESS, my_reporter))
or pass it directly to the worker bootstrap.
NullErrorReporter ¶
Default no-op :class:ErrorReporter — silently drops all reports.
Used when no vendor-specific error routing is configured. Instances are stateless and safe to share.
ProgressEvent ¶
Bases: BaseModel
Point-in-time progress snapshot published to Redis for SSE/stream fanout.
Covers both kind="progress" (incremental update) and
kind="state_change" (terminal or status transition) events. The
exclude_none=True flag on :meth:model_dump_json suppresses null
fields so the JSON payload stays compact on the wire.
Fail ¶
JobRetryState ¶
Bases: NamedTuple
Projection of JobRow columns consumed by the retry classifier.
schedule_to_close is no longer an input to classification — the
SQL deadline guard in mark_failed_or_retry is the single deadline
arbiter (one arbiter per predicate; see docs/architecture.md §Clock
Domains). The field is retained on the projection for observability
and hook consumers. start_to_close is reserved for per-attempt
timeout enforcement at the consumer level (asyncio.wait_for); not used
by the classifier.
Retry ¶
Bases: BaseModel
Retry decision: reschedule the job after retry_delay.
The delay — not a computed timestamp — is the decision payload: the
backend derives scheduled_at = now() + retry_delay and the
scheduled/pending status from its own clock (single arbiter, immune to
app↔DB clock skew). retry_delay=0 means "retry immediately"
(lands pending).
RetryClassifier ¶
Pure classifier that maps an exception + policy to a RetryDecision.
The classifier decides retry-kind and backoff only — it is
deliberately NOT a deadline arbiter. schedule_to_close is
arbitrated by the SQL guard in mark_failed_or_retry (single
arbiter, the backend's clock); a Python-side pre-check computed from
the worker's clock would disagree with it under app↔DB skew and kill
jobs early (or rubber-stamp them).
classify
staticmethod
¶
classify(
policy: RetryPolicy,
non_retryable_exceptions: tuple[
type[BaseException], ...
],
exception: BaseException,
attempt: int,
*,
max_retry_backoff: timedelta = DEFAULT_MAX_RETRY_BACKOFF,
override: RetryOverride | None = None,
) -> RetryDecision
Source code in src/taskq/retry.py
RetryOverride ¶
Bases: BaseModel
Per-exception override returned by an actor's retry_classifier hook.
Both fields are optional; None means "use the actor's static
RetryPolicy/computed backoff for this field." Returning a
RetryOverride with only kind set lets one exception type
branch into different retry behaviour per occurrence — e.g. an HTTP
429 response goes indefinite while a 404 response on the same
exception type goes non_retryable. Returning one with only
delay set lets the actor honour a server-provided retry-after
duration instead of the policy's computed exponential/linear
backoff, while max_retry_backoff still applies as a safety
ceiling so a malicious or malformed header cannot strand a job.
RetryPolicy ¶
Bases: BaseModel
Policy controlling retry behaviour for an actor.
backoff
class-attribute
instance-attribute
¶
OIDCSettings ¶
Bases: DotEnvConfig
OIDC SSO configuration (loaded from TASKQ_OIDC_* env vars).
issuer
class-attribute
instance-attribute
¶
issuer: str = Field(
default="",
description="OIDC discovery issuer URL (e.g. https://login.microsoftonline.com/{tenant}/v2.0).",
)
client_id
class-attribute
instance-attribute
¶
client_secret
class-attribute
instance-attribute
¶
redirect_uri
class-attribute
instance-attribute
¶
redirect_uri: str = Field(
default="",
description="Must match the app registration's configured redirect URI.",
)
session_secret
class-attribute
instance-attribute
¶
session_secret: str = Field(
default="",
description="Signing key for session cookies; use >=32 bytes of random data. Rotate to invalidate all sessions.",
)
session_max_age_seconds
class-attribute
instance-attribute
¶
session_max_age_seconds: int = Field(
default=28800,
ge=60,
description="Session lifetime (s). Default 8h.",
)
scope
class-attribute
instance-attribute
¶
scope: str = Field(
default="openid profile email",
description="OIDC scopes. Add 'Group.Read.All' for the Entra overage group_resolver (Graph API /me/memberOf).",
)
group_claim
class-attribute
instance-attribute
¶
group_claim: str | None = Field(
default=None,
description="ID token claim name for groups (e.g. 'groups', 'roles'). None = authentication-only authorization.",
)
allowed_groups
class-attribute
instance-attribute
¶
SAMLSettings ¶
Bases: DotEnvConfig
SAML SSO configuration (loaded from TASKQ_SAML_* env vars).
entity_id
class-attribute
instance-attribute
¶
acs_url
class-attribute
instance-attribute
¶
idp_entity_id
class-attribute
instance-attribute
¶
idp_sso_url
class-attribute
instance-attribute
¶
idp_x509_cert
class-attribute
instance-attribute
¶
sp_x509_cert
class-attribute
instance-attribute
¶
sp_x509_cert: str | None = Field(
default=None,
description="SP cert (signed requests / encrypted assertions).",
)
sp_private_key
class-attribute
instance-attribute
¶
session_secret
class-attribute
instance-attribute
¶
session_max_age_seconds
class-attribute
instance-attribute
¶
session_max_age_seconds: int = Field(
default=28800,
ge=60,
description="Session lifetime (s). Default 8h.",
)
group_attribute
class-attribute
instance-attribute
¶
allowed_groups
class-attribute
instance-attribute
¶
TaskQSettings ¶
Bases: DotEnvConfig
Top-level TaskQ runtime configuration.
pg_dsn
class-attribute
instance-attribute
¶
pg_dsn: PostgresDsn = Field(
default=PostgresDsn(
"postgresql://taskq:taskq@localhost:5432/taskq"
),
description="Direct (non-PgBouncer) DSN. LISTEN/NOTIFY and advisory locks need a session.",
)
schema_name
class-attribute
instance-attribute
¶
schema_name: str = Field(
default="taskq",
validator=_schema_name_validator,
description="Postgres schema for all TaskQ tables.",
)
redis_url
class-attribute
instance-attribute
¶
redis_url: RedisDsn | None = Field(
default=None,
description="Optional Redis URL. Required for real-time progress fanout.",
)
environment
class-attribute
instance-attribute
¶
environment: str | None = Field(
default=None,
description="TASKQ_ENVIRONMENT. Deployment environment label. The unauthenticated-admin WARNING ('admin-ui-no-auth') fires in EVERY environment whenever the admin UI is served without auth_dependency. 'dev' and 'development' additionally skip the fail-closed RuntimeError (the WARNING then notes the absence is the dev exemption); any other value (or None/empty) fails closed when admin_ui_require_auth is True.",
)
admin_max_sse_connections
class-attribute
instance-attribute
¶
admin_max_sse_connections: int = Field(
default=50,
ge=1,
description="TASKQ_ADMIN_MAX_SSE_CONNECTIONS. Maximum concurrent SSE connections the admin UI will serve. Used to size the connection-limit semaphore.",
)
progress_max_sse_connections
class-attribute
instance-attribute
¶
progress_max_sse_connections: int = Field(
default=50,
ge=1,
description="TASKQ_PROGRESS_MAX_SSE_CONNECTIONS. Maximum concurrent per-job progress SSE streams this process will serve. Each holds a Redis pubsub subscription and an asyncio task for as long as the client stays connected, so an uncapped endpoint is a resource-exhaustion surface on the app hosting the pipeline.",
)
admin_host
class-attribute
instance-attribute
¶
admin_host: str = Field(
default="0.0.0.0",
description="TASKQ_ADMIN_HOST. Bind address for ``taskq ui serve``.",
)
admin_port
class-attribute
instance-attribute
¶
admin_port: int = Field(
default=8080,
ge=1,
le=65535,
description="TASKQ_ADMIN_PORT. Bind port for ``taskq ui serve``.",
)
admin_url
class-attribute
instance-attribute
¶
admin_url: str = Field(
default="http://localhost:8080",
description="TASKQ_ADMIN_URL. Public base URL of the admin UI as seen from a browser. Used by the example trigger app to construct redirect URLs after enqueueing. In a shared-container deployment this is the external address of the admin process (e.g. http://localhost:8001). Override when admin and trigger app are on different hosts or ports.",
)
admin_ui_polling_interval_seconds
class-attribute
instance-attribute
¶
admin_ui_polling_interval_seconds: float = Field(
default=2.0,
ge=0.1,
description="TASKQ_ADMIN_UI_POLLING_INTERVAL_SECONDS. How often the admin UI polls PG in polling/degraded mode. Injected as poll_interval_ms into every template.",
)
admin_worker_liveness_seconds
class-attribute
instance-attribute
¶
admin_worker_liveness_seconds: int = Field(
default=30,
ge=1,
description="TASKQ_ADMIN_WORKER_LIVENESS_SECONDS. How recently a worker must have written last_seen_at to count as alive in the admin UI: it drives the 'queue has pending jobs but no alive worker' banner and the leader's watchdog_healthy verdict. Must comfortably exceed TASKQ_HEARTBEAT_INTERVAL (default 10 s), so the default 30 s is three beats; a deployment that lengthens the heartbeat, or whose PG is cross-region, has to raise this or every healthy worker reads as dead. Measured by Postgres against clock_timestamp(), never by the admin process's own clock.",
)
admin_ui_allow_rate_limit_reset
class-attribute
instance-attribute
¶
admin_ui_allow_rate_limit_reset: bool = Field(
default=False,
description="TASKQ_ADMIN_UI_ALLOW_RATE_LIMIT_RESET. When True, the admin UI shows a reset button on the rate-limits page and serves the POST /rate-limits/{bucket_name}/reset endpoint. Default False for safety - prevents accidental resets in production.",
)
admin_ui_require_auth
class-attribute
instance-attribute
¶
admin_ui_require_auth: bool = Field(
default=True,
description="TASKQ_ADMIN_UI_REQUIRE_AUTH. When True (the default), create_router raises RuntimeError if auth_dependency is None in a non-dev environment, failing closed. Set to False to suppress the error and allow an unauthenticated admin UI in non-dev (not recommended - only for air-gapped or localhost-only deployments).",
)
admin_ui_frame_ancestors
class-attribute
instance-attribute
¶
admin_ui_frame_ancestors: str = Field(
default="none",
validator=_frame_ancestors_validator,
description="TASKQ_ADMIN_UI_FRAME_ANCESTORS. Who may frame admin pages: 'none' (the default, nobody) or 'self' (the admin UI's own origin, for a host app that embeds the admin UI in its own dashboard). Emitted as both 'Content-Security-Policy: frame-ancestors ...' and the legacy 'X-Frame-Options' (DENY / SAMEORIGIN). CSRF is no defence against UI redress: the framed page is the real, authenticated, same-origin page, so a tricked click carries a valid token.",
)
admin_ui_secure_cookies
class-attribute
instance-attribute
¶
admin_ui_secure_cookies: bool = Field(
default=True,
description="TASKQ_ADMIN_UI_SECURE_COOKIES. Sets the 'Secure' flag on the admin UI's CSRF cookie. A configured value, not one inferred from request.url.scheme: behind a TLS-terminating edge (Azure Application Gateway, App Service) the app sees plain http, so an inferred flag is silently dropped on a connection the browser reached over HTTPS. Set False only for local http dev, where a Secure cookie is rejected by the browser and the admin UI stops working.",
)
admin_actions_enabled
class-attribute
instance-attribute
¶
admin_actions_enabled: bool = Field(
default=False,
description="TASKQ_ADMIN_ACTIONS_ENABLED. When True, the admin UI permits state-changing actions: run schedule now, enable/disable/skip a schedule, retry job, cancel job. Default False - prevents on-demand triggering of registered business logic, and silent suppression of scheduled work, via the admin UI without explicit opt-in. Separate from auth_dependency, which controls read access to all admin routes.",
)
pg_credential_provider
class-attribute
instance-attribute
¶
pg_credential_provider: str | None = Field(
default=None,
description="TASKQ_PG_CREDENTIAL_PROVIDER. Module:attr reference to a PgCredentialProvider (e.g. myapp.auth:make_provider) - an instance, a zero-arg factory returning one, or the provider class. Every Postgres pool and dedicated connection is then built through it, so SIGHUP / TASKQ_RELOAD_INTERVAL rotate real credentials. The CLI options --pg-credential-provider (taskq worker / migrate / ui serve) override it. The ref is resolved, not validated, at load time: the import lives in the CLI so a bad ref exits 1 with a pointed message instead of a settings traceback. See docs/guides/managed-identities.md.",
)
redis_credential_provider
class-attribute
instance-attribute
¶
redis_credential_provider: str | None = Field(
default=None,
description="TASKQ_REDIS_CREDENTIAL_PROVIDER. Module:attr reference to a RedisCredentialProvider, in the same shapes as pg_credential_provider. Requires TASKQ_REDIS_URL. Overridden by --redis-credential-provider.",
)
sso_backend
class-attribute
instance-attribute
¶
sso_backend: str = Field(
default="none",
validator=_sso_backend_validator,
description="TASKQ_SSO_BACKEND. Selects the SSO backend for the admin UI: 'none' (default, unauthenticated/BYO-auth), 'oidc' (taskq[oidc]), or 'saml' (taskq[saml]). See docs/guides/sso.md.",
)
health_token
class-attribute
instance-attribute
¶
health_token: str = Field(
default="",
description="TASKQ_HEALTH_TOKEN. Bearer token for machine-to-machine access to health/metrics endpoints. When set, health and metrics routes require a matching 'Authorization: Bearer <token>' header. Leave empty for unauthenticated cluster-internal access - but see health_require_token, which fails closed on an empty token outside dev.",
)
health_require_token
class-attribute
instance-attribute
¶
health_require_token: bool = Field(
default=True,
description="TASKQ_HEALTH_REQUIRE_TOKEN. When True (the default), taskq ui serve raises RuntimeError if health_token is empty in a non-dev environment, failing closed. Set to False to suppress the error and allow unauthenticated health/metrics endpoints in non-dev (e.g. when relying on network policy / cluster-internal-only access instead of a bearer token - note that many k8s liveness/readiness probes don't send auth headers by default, so enabling the token may require updating the probe config too).",
)
migrate_on_start
class-attribute
instance-attribute
¶
migrate_on_start: bool = Field(
default=False,
description="TASKQ_MIGRATE_ON_START. When True, apply pending migrations before the admin UI accepts its first request. Aborts startup if migrations fail. Consumed ONLY by `taskq ui serve` -- the worker ignores it (and warns when it is set), because N worker replicas racing to migrate is the concurrent-migration hazard migrations are supposed to avoid. Migrate from a pre-deploy job or init container.",
)
example_host
class-attribute
instance-attribute
¶
example_host: str = Field(
default="0.0.0.0",
description="TASKQ_EXAMPLE_HOST. Bind address for the example trigger app (uvicorn). Only consumed by the example app; ignored by the worker and admin UI.",
)
example_port
class-attribute
instance-attribute
¶
example_port: int = Field(
default=8000,
ge=1,
le=65535,
description="TASKQ_EXAMPLE_PORT. Bind port for the example trigger app (uvicorn). Only consumed by the example app; ignored by the worker and admin UI.",
)
idempotency_key_max_bytes
class-attribute
instance-attribute
¶
idempotency_key_max_bytes: int = Field(
default=MAX_IDEMPOTENCY_KEY_BYTES,
ge=1,
le=IDEMPOTENCY_KEY_BYTES_CEILING,
description="TASKQ_IDEMPOTENCY_KEY_MAX_BYTES. Maximum UTF-8 byte length of idempotency_key, and of idempotency_scope, each. Bytes rather than characters because the real bound is the composite unique index jobs_idempotency_scope_key_uniq: a btree v4 entry cannot exceed 2704 bytes, counted encoded. The ceiling keeps scope + key + index-tuple overhead under that, so no value here can turn a valid enqueue into a raw Postgres 'index row size ... exceeds btree version 4 maximum'. Raise it when keys are derived from URLs, composite business keys or opaque vendor cursors.",
)
oidc
property
¶
Lazily loaded OIDC sub-config (TASKQ_OIDC_* env vars).
Backed by dotenvmodel's cached() singleton: the environment is
read on first access and the same instance returned thereafter.
Reload (e.g. on SIGHUP): call OIDCSettings.cached().reload() to
re-read the environment and mutate the shared instance in place -
every holder observes the new values - or
OIDCSettings.reset_cached() to force the next access to
re-load. Tests that change TASKQ_OIDC_* mid-process must do
the same (or use cached_override()).
saml
property
¶
Lazily loaded SAML sub-config (TASKQ_SAML_* env vars).
See :attr:oidc for the singleton/caching semantics and the
SIGHUP-style reload recipe (SAMLSettings.cached().reload() /
SAMLSettings.reset_cached()).
load
classmethod
¶
load(
env: str | None = None,
*,
override: bool | None = None,
env_dir: Path | str | None = None,
read_dotfiles: bool | None = None,
read_environ: bool | None = None,
load_local: bool | None = None,
) -> Self
Load settings via dotenvmodel's cascading .env discovery.
All parameters are forwarded to DotEnvConfig.load unchanged
(resolution: explicit argument > DOTENV_* env var > default).
override=None keeps dotenvmodel's default precedence — the
process environment beats .env files; pass override=True
or set DOTENV_OVERRIDE=true to make .env files win instead.
read_dotfiles=False / read_environ=False disable the
.env cascade / the process environment respectively, per
dotenvmodel's documented symmetry.
dotenvmodel logs a WARNING ("No .env files found in .env file is present - noisy on every CLI
invocation in projects that configure purely via real environment
variables. read_dotfiles=False is not the answer: it silences
the warning but disables the .env cascade entirely, a
documented core TaskQ feature. This override instead installs a
:class:_NoEnvFilesWarningFilter on the dotenvmodel logger
for the duration of the call, dropping only that one warning;
every other dotenvmodel warning (e.g. an invalid DOTENV_*
knob value) remains visible.
WorkerSettings.load inherits this override via MRO, so worker
startup gets the same quiet load.
Source code in src/taskq/settings.py
WorkerSettings ¶
Bases: TaskQSettings
Worker-specific configuration with three-pool sizing and dual-DSN support.
Extends :class:TaskQSettings with pool-size knobs, dual-DSN fields, and
the validated lock_lease >= 4 * heartbeat_interval invariant.
pg_dsn_direct
class-attribute
instance-attribute
¶
pg_dsn_direct: PostgresDsn | None = Field(
default=None,
description="TASKQ_PG_DSN_DIRECT; falls back to pg_dsn when absent. Bypasses PgBouncer - used by dispatcher_pool, heartbeat_pool, notify_conn, and leader_conn.",
)
pg_dsn_pooled
class-attribute
instance-attribute
¶
pg_dsn_pooled: PostgresDsn | None = Field(
default=None,
description="TASKQ_PG_DSN_POOLED; falls back to pg_dsn when absent. May route through PgBouncer transaction mode - used by worker_pool only.",
)
dispatcher_pool_size
class-attribute
instance-attribute
¶
dispatcher_pool_size: int = Field(
default=4,
ge=1,
description="TASKQ_DISPATCHER_POOL_SIZE. Max connections for the dispatcher pool. Bypasses PgBouncer.",
)
dispatcher_command_timeout
class-attribute
instance-attribute
¶
dispatcher_command_timeout: float = Field(
default=5.0,
ge=1.0,
description="TASKQ_DISPATCHER_COMMAND_TIMEOUT (seconds). Per-query timeout for the dispatcher pool and the TaskQ-built leader connections (election, cron, monitor), and the single deadline wrapped around each period-1 leader-loop iteration (scheduled_wake, cron): a stalled PG errors the iteration instead of hanging the loop past its staleness budget. Checked at load time when the watchdog is enabled: timeout + the 1.0s leader-loop period must be < max(period x watchdog_tick_grace_factor, watchdog_stale_floor) for the period-1 leader loops (scheduled_wake, cron), so a timeout-capped iteration can never false-trip the stale-loop detector on a healthy worker. The producer loop is not checked (its multi-statement dispatch_batch is not wrapped in a single asyncio.timeout).",
)
dispatch_oversample
class-attribute
instance-attribute
¶
dispatch_oversample: int = Field(
default=2,
ge=1,
le=1000,
description="TASKQ_DISPATCH_OVERSAMPLE. Multiplier for per-actor candidate gathering in the dispatch SQL. Each LATERAL reads residual x oversample candidates. Higher values absorb more identity collisions and multi-producer contention. Default 2 (tolerates 50% dupe identities). Set 1 when no identity_key is used and single-producer.",
)
dispatch_scope_by_home_queue
class-attribute
instance-attribute
¶
dispatch_scope_by_home_queue: bool = Field(
default=False,
description="TASKQ_DISPATCH_SCOPE_BY_HOME_QUEUE. When True, restrict per_actor_capacity to actors whose home queue (actor_config.queue) the worker subscribes to. Lowers per-cycle probe count at the cost of not dispatching enqueue(queue=...) override jobs whose actor's home queue is not subscribed. Default False (override-safe).",
)
heartbeat_pool_size
class-attribute
instance-attribute
¶
heartbeat_pool_size: int = Field(
default=4,
ge=1,
description="TASKQ_HEARTBEAT_POOL_SIZE. Max connections for the heartbeat pool. Bypasses PgBouncer.",
)
heartbeat_command_timeout
class-attribute
instance-attribute
¶
heartbeat_command_timeout: float = Field(
default=2.0,
gt=0.0,
description="TASKQ_HEARTBEAT_COMMAND_TIMEOUT (seconds). Per-query timeout for the heartbeat pool, deliberately tighter than dispatcher_command_timeout: a beat that takes longer than the tick cannot keep a lock lease alive, so failing it fast is the point. Raise it on a loaded or cross-region Postgres — max_heartbeat_failures consecutive timeouts self-terminate the worker, and before this was a setting the 2 s literal made that unavoidable. Must be > 0: asyncpg reads 0 as 'no timeout', which turns a stalled beat into a hang.",
)
max_concurrency
class-attribute
instance-attribute
¶
max_concurrency: int = Field(
default=8,
ge=1,
description="TASKQ_MAX_CONCURRENCY. Upper bound on concurrent jobs. worker_pool max_size = int(max_concurrency * 1.5).",
)
heartbeat_interval
class-attribute
instance-attribute
¶
heartbeat_interval: float = Field(
default=10.0,
ge=0.5,
description="TASKQ_HEARTBEAT_INTERVAL (seconds). Period between heartbeat ticks.",
)
lock_lease
class-attribute
instance-attribute
¶
lock_lease: float = Field(
default=60.0,
ge=1.0,
description="TASKQ_LOCK_LEASE (seconds). Time before a held lock is reclaimed by the recovery sweep. Must be >= 4 * heartbeat_interval.",
)
max_heartbeat_failures
class-attribute
instance-attribute
¶
max_heartbeat_failures: int = Field(
default=3,
ge=1,
description="TASKQ_MAX_HEARTBEAT_FAILURES. Consecutive heartbeat failures before the worker self-terminates.",
)
sweep_interval
class-attribute
instance-attribute
¶
sweep_interval: float = Field(
default=30.0,
ge=1.0,
description="TASKQ_SWEEP_INTERVAL (seconds). Period between leader sweep loop iterations — reclaim_expired_locks, sweep_expired_results, cleanup_stale_workers, and idle keyed-ref eviction. Lower values reduce recovery latency for crashed workers at the cost of more frequent PG queries.",
)
queue_depth_interval
class-attribute
instance-attribute
¶
queue_depth_interval: float = Field(
default=15.0,
ge=1.0,
description="TASKQ_QUEUE_DEPTH_INTERVAL (seconds). Period between queue-depth metrics sampling iterations.",
)
reservation_slots_interval
class-attribute
instance-attribute
¶
reservation_slots_interval: float = Field(
default=15.0,
ge=1.0,
description="TASKQ_RESERVATION_SLOTS_INTERVAL (seconds). Period between reservation-slot metrics sampling iterations.",
)
stranded_jobs_interval
class-attribute
instance-attribute
¶
stranded_jobs_interval: float = Field(
default=60.0,
ge=1.0,
description="TASKQ_STRANDED_JOBS_INTERVAL (seconds). Period between stranded-jobs (pending jobs whose actor has no actor_config) warning checks.",
)
termination_grace_period
class-attribute
instance-attribute
¶
termination_grace_period: float = Field(
default=75.0,
ge=5.0,
description="TASKQ_TERMINATION_GRACE_PERIOD (seconds). Total wall-clock budget from SIGTERM to forced exit; the shutdown watchdog counts it down from the first shutdown signal. Must satisfy cancellation_grace + cleanup_grace < termination_grace - 5, and should cover the modelled worst case cancellation_grace + cleanup_grace + the ~27s bounded-close teardown tail (see WorkerSettings.worst_case_shutdown_seconds) — the default does: 30 + 10 + 27 = 67s, with headroom for the ~72s sibling-crash path that model understates. A value below the worst case still loads but logs shutdown-budget-exceeds-termination-grace at startup. Size the pod/container grace (terminationGracePeriodSeconds / stop_grace_period) from the same worst case, not from this setting alone.",
)
cancellation_grace_period
class-attribute
instance-attribute
¶
cancellation_grace_period: float = Field(
default=30.0,
ge=0.0,
description="TASKQ_CANCELLATION_GRACE_PERIOD (seconds). Cooperative cancel phase duration.",
)
cleanup_grace_period
class-attribute
instance-attribute
¶
cleanup_grace_period: float = Field(
default=10.0,
ge=0.0,
description="TASKQ_CLEANUP_GRACE_PERIOD (seconds). Force-cancel cleanup grace.",
)
reclaim_event_visibility_delay
class-attribute
instance-attribute
¶
reclaim_event_visibility_delay: float = Field(
default=RECLAIM_EVENT_VISIBILITY_DELAY.total_seconds(),
ge=0.0,
description="TASKQ_RECLAIM_EVENT_VISIBILITY_DELAY (seconds). Trailing-watermark margin poll_reclaim_events()/TaskQ.watch_reclaims() apply before returning a job_events row, so an out-of-commit-order sibling with a lower event_id has time to appear first (see docs/architecture.md's crash-reclaim section). Correctness assumes every job_events writer transaction commits within this margin of its INSERT; raise it if sweeps run under heavy lock contention or against very large batches, lower it if latency matters more and writes are known to be fast. A writer that exceeds the margin can cause a silently missed event - this is a real, not merely theoretical, risk under misconfiguration.",
)
max_retry_backoff
class-attribute
instance-attribute
¶
max_retry_backoff: timedelta = Field(
default=DEFAULT_MAX_RETRY_BACKOFF,
description="TASKQ_MAX_RETRY_BACKOFF (interval). Global ceiling on retry backoff per attempt - caps the per-actor RetryPolicy.cap so a misconfigured actor (e.g. cap=timedelta(days=365)) cannot strand jobs for an unreasonably long time. Default 24 h: conservative, matches one standard on-call rotation, and mirrors Dramatiq's DEFAULT_MAX_BACKOFF philosophy ",
)
default_start_to_close
class-attribute
instance-attribute
¶
default_start_to_close: timedelta | None = Field(
default=None,
validator=_positive_timedelta,
description="TASKQ_DEFAULT_START_TO_CLOSE (interval). Worker-side fallback per-attempt execution timeout, applied only when a job has no start_to_close of its own (neither passed at enqueue time nor declared as an @actor(start_to_close=...) default). None (the default) means unbounded - matches existing behaviour, opt-in only. Set this to give every actor on this worker a safety-net wall-clock budget per attempt, preventing a hung or infinite-looping actor from occupying a coroutine slot forever, without having to configure start_to_close on every individual actor. Precedence (highest wins): per-enqueue start_to_close > @actor(start_to_close=...) > this setting. This does not affect schedule_to_close, which is a separate, unrelated deadline for the job's *overall* retry budget across all attempts - start_to_close bounds a single attempt's wall-clock time.",
)
rate_limit_pg_fallback_enabled
class-attribute
instance-attribute
¶
rate_limit_pg_fallback_enabled: bool = Field(
default=True,
description="TASKQ_RATE_LIMIT_PG_FALLBACK_ENABLED. When False, Redis errors propagate instead of triggering PG fallback.",
)
max_keyed_reservations
class-attribute
instance-attribute
¶
max_keyed_reservations: int = Field(
default=10000,
ge=1,
description="TASKQ_MAX_KEYED_RESERVATIONS. Guardrail on the number of distinct keyed-reservation entries tracked in memory. When the limit is reached, new keyed reservations raise ReservationUnavailable. Tune to your workload's expected key cardinality.",
)
max_keyed_rate_limits
class-attribute
instance-attribute
¶
max_keyed_rate_limits: int = Field(
default=10000,
ge=1,
description="TASKQ_MAX_KEYED_RATE_LIMITS. Guardrail on the number of distinct keyed-rate-limit entries tracked in memory. When the limit is reached, new keyed rate limits raise ReservationUnavailable. Independent from max_keyed_reservations, which governs keyed reservations only. Tune to your workload's expected key cardinality.",
)
metrics_port
class-attribute
instance-attribute
¶
metrics_port: int = Field(
default=9090,
ge=1,
le=65535,
description="TASKQ_METRICS_PORT. Bind port for the standalone Prometheus metrics server (taskq health metrics --port). The in-process FastAPI mount ignores this field.",
)
health_enabled
class-attribute
instance-attribute
¶
health_enabled: bool = Field(
default=True,
description="TASKQ_HEALTH_ENABLED. Enable the Unix-socket health server.",
)
health_socket_path
class-attribute
instance-attribute
¶
health_socket_path: str = Field(
default="/tmp/taskq_health.sock",
description="TASKQ_HEALTH_SOCKET_PATH. Unix socket path for the health server.",
)
health_pg_ping_timeout
class-attribute
instance-attribute
¶
health_pg_ping_timeout: float = Field(
default=0.2,
ge=0.0,
description="TASKQ_HEALTH_PG_PING_TIMEOUT. Seconds to wait for dispatcher_pool.acquire() in the readiness PG ping. Default 200ms .",
)
health_tasks_enabled
class-attribute
instance-attribute
¶
health_tasks_enabled: bool = Field(
default=False,
description="TASKQ_HEALTH_TASKS_ENABLED. Expose the privileged /tasks asyncio stack-dump endpoint on the Unix health socket. Off by default: the dump reveals code structure, file paths, and task names (never locals or payload values). Enabling it also tightens the socket to owner-only (no group/other access). Unix socket only — never mounted on the admin UI surface.",
)
health_host
class-attribute
instance-attribute
¶
health_host: str = Field(
default="0.0.0.0",
description="TASKQ_HEALTH_HOST. Bind address for the optional TCP health listener. Only used when health_port is set. Defaults to all interfaces because Azure Container Apps and Kubernetes probe the replica over the pod network; narrow it to 127.0.0.1 when a local sidecar is the only prober.",
)
health_port
class-attribute
instance-attribute
¶
health_port: int | None = Field(
default=None,
ge=0,
le=65535,
description="TASKQ_HEALTH_PORT. TCP port for the HTTP health listener serving /live and /ready. Unset (the default) means no TCP listener at all — setting a port is the opt-in. Required on Azure Container Apps, whose probes support only httpGet/tcpSocket and cannot reach a Unix socket (there is no exec probe type). The Unix socket keeps working either way. If the port cannot be bound the worker fails to start rather than run with probes silently dead. 0 binds an ephemeral port (tests only).",
)
health_request_timeout
class-attribute
instance-attribute
¶
health_request_timeout: float = Field(
default=2.0,
gt=0.0,
description="TASKQ_HEALTH_REQUEST_TIMEOUT. Seconds allowed for a probe to send its whole request line and headers before the connection is dropped unanswered. Bounds a drip-feed client that would otherwise hold a connection open forever by staying just inside a per-line timeout. Keep it at or below the shortest probe timeoutSeconds you configure.",
)
health_max_header_bytes
class-attribute
instance-attribute
¶
health_max_header_bytes: int = Field(
default=16 * 1024,
gt=0,
description="TASKQ_HEALTH_MAX_HEADER_BYTES. Cap on a probe request's accumulated request line plus headers. Pairs with health_request_timeout to bound a peer that sends many small lines fast enough to stay inside the deadline. 16 KiB is far above any real probe request, which carries a path and a handful of headers.",
)
health_readiness_check_timeout
class-attribute
instance-attribute
¶
health_readiness_check_timeout: float = Field(
default=5.0,
gt=0.0,
description="TASKQ_HEALTH_READINESS_CHECK_TIMEOUT. Seconds each check registered via taskq.worker.health.register_readiness_check may take before it counts as a readiness failure. Defaults to 5s, matching the Azure Container Apps default readiness probe timeoutSeconds, so a wedged check fails the probe rather than outliving it.",
)
watchdog_enabled
class-attribute
instance-attribute
¶
watchdog_enabled: bool = Field(
default=True,
description="TASKQ_WATCHDOG_ENABLED. Master switch for the in-worker watchdog detectors (shutdown deadline, stale loop ticks, sibling contract, event-loop lag). A detector trip dumps the asyncio task stacks and force-exits non-zero so the supervisor restarts the worker instead of leaving it wedged.",
)
watchdog_loop_lag_budget
class-attribute
instance-attribute
¶
watchdog_loop_lag_budget: float = Field(
default=30.0,
gt=0.0,
description="TASKQ_WATCHDOG_LOOP_LAG_BUDGET (seconds). How long the event loop may go without scheduling before the lag watchdog trips. Deliberately far beyond any legitimate pause (GC, a slow tick) because the trip is terminal. Tier 2 of the lag detector; see watchdog_loop_lag_warn_budget for the non-terminal tier 1.",
)
watchdog_loop_lag_warn_budget
class-attribute
instance-attribute
¶
watchdog_loop_lag_warn_budget: float = Field(
default=5.0,
gt=0.0,
description="TASKQ_WATCHDOG_LOOP_LAG_WARN_BUDGET (seconds). Non-terminal tier-1 event-loop lag threshold: faulthandler thread dump + metric + deferred asyncio task-stack dump. Never exits; the terminal tier is watchdog_loop_lag_budget.",
)
watchdog_loop_lag_startup_grace
class-attribute
instance-attribute
¶
watchdog_loop_lag_startup_grace: float = Field(
default=30.0,
ge=0.0,
description="TASKQ_WATCHDOG_LOOP_LAG_STARTUP_GRACE (seconds). Grace before the lag watchdog arms, covering import-heavy startup, DI bootstrap, and first dispatch. Anchored to thread start; the lag detector also arms early once the first loop liveness tick lands.",
)
watchdog_tick_grace_factor
class-attribute
instance-attribute
¶
watchdog_tick_grace_factor: float = Field(
default=5.0,
gt=0.0,
description="TASKQ_WATCHDOG_TICK_GRACE_FACTOR. Multiplier on a loop's iteration period before its liveness tick is declared stale (floor 10s). Generous on purpose: a terminal detector must never fire on a merely loaded host.",
)
watchdog_dump_interval
class-attribute
instance-attribute
¶
watchdog_dump_interval: float = Field(
default=5.0,
gt=0.0,
description="TASKQ_WATCHDOG_DUMP_INTERVAL (seconds). Interval between straggler logs (names + await sites of still-alive siblings) while a shutdown is in progress.",
)
watchdog_dump_after_fraction
class-attribute
instance-attribute
¶
watchdog_dump_after_fraction: float = Field(
default=0.5,
gt=0.0,
lt=1.0,
description="TASKQ_WATCHDOG_DUMP_AFTER_FRACTION. Fraction of the shutdown deadline that must be consumed before straggler dumps begin (0.5 = only in the back half of the budget). A drain inside its front half is within expectations and stays quiet; one countdown-start record is always logged so the window is never blind. Must be < 1: at 1.0 the deadline trip would always fire first, silently disabling the dumps.",
)
watchdog_stale_floor
class-attribute
instance-attribute
¶
watchdog_stale_floor: float = Field(
default=10.0,
gt=0.0,
description="TASKQ_WATCHDOG_STALE_FLOOR (seconds). Minimum staleness budget for any loop (period x grace_factor, floored at this value). Guards tiny intervals against false trips under host starvation — a terminal detector must never fire on load.",
)
watchdog_check_interval
class-attribute
instance-attribute
¶
watchdog_check_interval: float = Field(
default=1.0,
gt=0.0,
description="TASKQ_WATCHDOG_CHECK_INTERVAL (seconds). Poll cadence for the stale-tick sweep and the loop-lag watchdog thread.",
)
poll_interval
class-attribute
instance-attribute
¶
poll_interval: float = Field(
default=1.0,
gt=0,
description="TASKQ_POLL_INTERVAL (seconds). Producer loop fallback polling cadence when the NOTIFY listener is unavailable.",
)
notify_health_check_interval
class-attribute
instance-attribute
¶
notify_health_check_interval: float = Field(
default=5.0,
gt=0,
description="TASKQ_NOTIFY_HEALTH_CHECK_INTERVAL (seconds). How often _health_check_loop issues SELECT 1 on notify_conn. Detection latency before reconnect is at most this interval.",
)
notify_reconnect_backoff_initial
class-attribute
instance-attribute
¶
notify_reconnect_backoff_initial: float = Field(
default=1.0,
gt=0,
description="TASKQ_NOTIFY_RECONNECT_BACKOFF_INITIAL (seconds). Initial exponential backoff delay before the first reconnect retry. Cap is 30 s (factor 2 per attempt). Backoff sequence: 1, 2, 4, 8, 16, 30.",
)
notify_listener_setup_timeout
class-attribute
instance-attribute
¶
notify_listener_setup_timeout: float = Field(
default=10.0,
validator=_positive_finite_float,
description="TASKQ_NOTIFY_LISTENER_SETUP_TIMEOUT (seconds). Bounds each ``add_listener`` call during NOTIFY listener setup and reconnect - a half-open PG connection that accepts TCP but stalls on the LISTEN handshake would otherwise wedge the notify loop forever. On timeout the connection is closed (bounded) and the reconnect retry loop is entered (or the initial setup raises).",
)
notify_enabled
class-attribute
instance-attribute
¶
notify_enabled: bool = Field(
default=True,
description="TASKQ_NOTIFY_ENABLED. When True, the worker uses LISTEN/NOTIFY for near-zero-latency dispatch wakeups with poll interval as fallback. When False, the worker uses poll-only dispatch.",
)
notify_poll_interval
class-attribute
instance-attribute
¶
notify_poll_interval: float = Field(
default=5.0,
ge=0.5,
description="TASKQ_NOTIFY_POLL_INTERVAL (seconds). Fallback poll cadence when NOTIFY is enabled (rarely reached - NOTIFY handles the common case). Use poll_interval when NOTIFY is disabled.",
)
reload_interval
class-attribute
instance-attribute
¶
reload_interval: float | None = Field(
default=None,
gt=0,
description="TASKQ_RELOAD_INTERVAL (seconds). When set, the worker periodically triggers a credential hot-reload (the same path as SIGHUP) with no external signal required - the rotation path for platforms without SIGHUP (e.g. Windows) and for hands-off scheduled rotation (e.g. ~720s for AWS IAM's 15-minute tokens). None disables the timer; SIGHUP and deps.request_reload() still work. Only factory-backed resources are rebuilt; DSN/static credentials are unaffected.",
)
reload_factory_timeout
class-attribute
instance-attribute
¶
reload_factory_timeout: float = Field(
default=30.0,
gt=0,
description="TASKQ_RELOAD_FACTORY_TIMEOUT (seconds). Bounds each individual factory call during a credential hot-reload - a hung token endpoint is marked failed for that resource instead of wedging the reload coordinator (and all future SIGHUPs).",
)
queues
class-attribute
instance-attribute
¶
queues: list[str] = Field(
default_factory=lambda: ["default"],
validator=_queue_names_validator,
description="TASKQ_QUEUES. Comma-separated list of queue names this worker will consume from.",
)
worker_label
class-attribute
instance-attribute
¶
worker_label: str | None = Field(
default=None,
validator=_worker_label_validator,
description="TASKQ_WORKER_LABEL. Human-readable label stored in the workers table for correlation with workgroup supervisors and external monitoring. When omitted the column is NULL; hostname and pid columns provide identification.",
)
workgroup_instance
class-attribute
instance-attribute
¶
workgroup_instance: str | None = Field(
default=None,
validator=_workgroup_instance_validator,
description="TASKQ_WORKGROUP_INSTANCE. UUIDv7 identifying the workgroup orchestrator that launched this worker. Used for cross-process correlation.",
)
pool_max_inactive_lifetime
class-attribute
instance-attribute
¶
pool_max_inactive_lifetime: float = Field(
default=300.0,
ge=0.0,
description="TASKQ_POOL_MAX_INACTIVE_LIFETIME (seconds). asyncpg max_inactive_connection_lifetime - closes connections idle longer than this threshold. Set to 3600.0 to match a typical SQLAlchemy pool_recycle=3600 setting when running alongside an SQLAlchemy-based service. Applied to dispatcher_pool, heartbeat_pool, and worker_pool.",
)
otel_enabled
class-attribute
instance-attribute
¶
otel_enabled: bool = Field(
default=True,
description="TASKQ_OTEL_ENABLED. When False, the library suppresses all span and metric creation but operations still succeed .",
)
exception_message_max_chars
class-attribute
instance-attribute
¶
exception_message_max_chars: int = Field(
default=2000,
ge=100,
description="TASKQ_EXCEPTION_MESSAGE_MAX_CHARS. Bound on exception message text on spans and logs, after scrubbing. Matches the admin UI's traceback bound so there is one number for how much error text is kept, not two. Truncation appends the dropped character count, so an operator can see text was cut and raise this. Raise it when an actor formats large context into its messages; the stack trace is a separate field and is not bounded by this.",
)
exception_redaction_enabled
class-attribute
instance-attribute
¶
exception_redaction_enabled: bool = Field(
default=True,
description="TASKQ_EXCEPTION_REDACTION_ENABLED. When True (the default), Postgres 'DETAIL:' lines are dropped from exception text before it reaches spans and logs, because they quote caller-supplied row values (idempotency_key, identity_key, fairness_key routinely hold tenant or subject identifiers). Set to False for advanced debugging to ship the raw text, including those row values, to every configured telemetry backend; the worker logs a startup WARNING while it is off. URI credential masking (scheme://user:***@host) is NOT affected by this setting and is always applied - no debugging case justifies sending a password to a telemetry vendor.",
)
worker_group
class-attribute
instance-attribute
¶
worker_group: str = Field(
default="default",
description="TASKQ_WORKER_GROUP. Consumer group name emitted as messaging.consumer.group.name on CONSUMER spans .",
)
log_format
class-attribute
instance-attribute
¶
log_format: str = Field(
default="json",
validator=_log_format_validator,
description="TASKQ_LOG_FORMAT. json|console. Selects JSONRenderer or ConsoleRenderer in setup_logging.",
)
log_level
class-attribute
instance-attribute
¶
log_level: str = Field(
default="INFO",
validator=_log_level_validator,
description="TASKQ_LOG_LEVEL. Root logger level.",
)
prune_schedule_utc
class-attribute
instance-attribute
¶
prune_schedule_utc: str = Field(
default="03:00",
validator=_hh_mm_validator,
description="TASKQ_PRUNE_SCHEDULE_UTC. HH:MM (UTC) for the daily prune run. Ignored when prune_cron_expr is set.",
)
prune_cron_expr
class-attribute
instance-attribute
¶
prune_cron_expr: str | None = Field(
default=None,
validator=_cron_expr_validator,
description="TASKQ_PRUNE_CRON_EXPR. Full 5-field cron expression. When set, takes precedence over prune_schedule_utc.",
)
prune_batch_size
class-attribute
instance-attribute
¶
prune_batch_size: int = Field(
default=DEFAULT_PRUNE_BATCH_SIZE,
ge=1,
description="TASKQ_PRUNE_BATCH_SIZE. Rows to delete per batch.",
)
prune_retention_period
class-attribute
instance-attribute
¶
prune_retention_period: timedelta = Field(
default=DEFAULT_PRUNE_RETENTION,
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_PERIOD. Global fallback retention. timedelta(0) means archive all terminal jobs immediately (valid). Negative values raise ConstraintViolationError at settings load.",
)
prune_retention_succeeded
class-attribute
instance-attribute
¶
prune_retention_succeeded: timedelta = Field(
default=timedelta(days=30),
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_SUCCEEDED.",
)
prune_retention_failed
class-attribute
instance-attribute
¶
prune_retention_failed: timedelta = Field(
default=timedelta(days=90),
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_FAILED.",
)
prune_retention_cancelled
class-attribute
instance-attribute
¶
prune_retention_cancelled: timedelta = Field(
default=timedelta(days=30),
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_CANCELLED.",
)
prune_retention_abandoned
class-attribute
instance-attribute
¶
prune_retention_abandoned: timedelta = Field(
default=timedelta(days=90),
validator=_non_negative_timedelta,
description="TASKQ_PRUNE_RETENTION_ABANDONED. Also used for crashed jobs (no separate prune_retention_crashed field).",
)
archive_retention_period
class-attribute
instance-attribute
¶
archive_retention_period: timedelta = Field(
default=timedelta(days=365),
validator=_non_negative_timedelta,
description="TASKQ_ARCHIVE_RETENTION_PERIOD. How long archived jobs are retained in jobs_archive before hard-deletion. Default 1 year. timedelta(0) is valid. Negative values raise ConstraintViolationError.",
)
archive_expiry_schedule_utc
class-attribute
instance-attribute
¶
archive_expiry_schedule_utc: str = Field(
default="04:00",
validator=_hh_mm_validator,
description="TASKQ_ARCHIVE_EXPIRY_SCHEDULE_UTC. HH:MM (UTC) for the daily archive expiry sweep. Default 04:00, 1 hour after the prune sweep.",
)
archive_expiry_cron_expr
class-attribute
instance-attribute
¶
archive_expiry_cron_expr: str | None = Field(
default=None,
validator=_cron_expr_validator,
description="TASKQ_ARCHIVE_EXPIRY_CRON_EXPR. Full 5-field cron expression. When set, takes precedence over archive_expiry_schedule_utc.",
)
force_update_actor_config
class-attribute
instance-attribute
¶
force_update_actor_config: bool = Field(
default=False,
description="When True, sync_actor_config silently overwrites a stored actor_config row whose queue or metadata differ from the registered values. When False (the default), that structural drift raises ActorConfigDriftList and the worker refuses to start. Capacity fields (max_concurrent, max_pending, result_ttl) are unaffected by this flag: once a row exists, the stored value is always authoritative and is never overwritten by the registered @actor(...) literal, regardless of force. Use `taskq actor-config set` to change a stored capacity value. Env var: TASKQ_FORCE_UPDATE_ACTOR_CONFIG.",
)
progress_coalesce_interval
class-attribute
instance-attribute
¶
progress_coalesce_interval: float = Field(
default=0.5,
ge=0.1,
description="TASKQ_PROGRESS_COALESCE_INTERVAL (seconds). How long the periodic flush loop waits between writing coalesced progress state to Postgres. Redis publishes are not throttled by this setting - each ctx.progress() call publishes immediately (fire-and-forget). Lower values increase PG write frequency; minimum 0.1 s.",
)
progress_data_max_bytes
class-attribute
instance-attribute
¶
progress_data_max_bytes: int = Field(
default=16384,
ge=1024,
le=1048576,
description="TASKQ_PROGRESS_DATA_MAX_BYTES. Maximum serialised byte length of the ``data`` dict in a single progress call. Payloads exceeding this limit raise ProgressTooLarge . Range: 1 KiB - 1 MiB; default 16 KiB.",
)
progress_publish_global
class-attribute
instance-attribute
¶
progress_publish_global: bool = Field(
default=True,
description="TASKQ_PROGRESS_PUBLISH_GLOBAL. When True (the default), progress events are additionally published to a schema-wide global fanout channel (in addition to the per-job channel). When False, events are only published to the per-job Redis channel. Does not affect Postgres flushing.",
)
result_max_bytes
class-attribute
instance-attribute
¶
result_max_bytes: int = Field(
default=MAX_RESULT_BYTES,
ge=1024,
le=1048576,
description="TASKQ_RESULT_MAX_BYTES. Maximum serialised byte length of a job's terminal result dict. A larger result raises ResultTooLarge, which is non-retryable — the actor already ran, so a re-run returns the same oversized value. Range: 1 KiB - 1 MiB (the same ceiling as progress_data_max_bytes, so the durable payload can be configured as large as the transient one); default 64 KiB. Raise it only with the row size in mind: unlike progress data, the result is stored for the job's result_ttl.",
)
cron_catch_up_window
class-attribute
instance-attribute
¶
cron_catch_up_window: timedelta = Field(
default=timedelta(hours=1),
validator=_non_negative_timedelta,
description="TASKQ_CRON_CATCH_UP_WINDOW. Missed firings within this window are caught up sequentially; older misses are skipped.",
)
cron_auto_disable_threshold
class-attribute
instance-attribute
¶
cron_auto_disable_threshold: int = Field(
default=3,
ge=1,
description="TASKQ_CRON_AUTO_DISABLE_THRESHOLD. Consecutive failures before a schedule is auto-disabled.",
)
idle_settle_window
class-attribute
instance-attribute
¶
idle_settle_window: float = Field(
default=2.0,
ge=0.0,
description="TASKQ_IDLE_SETTLE_WINDOW (seconds). Time the drain monitor waits after queues appear empty before declaring drained. Only used when --until-idle is active.",
)
idle_poll_interval
class-attribute
instance-attribute
¶
idle_poll_interval: float = Field(
default=1.0,
ge=0.1,
description="TASKQ_IDLE_POLL_INTERVAL (seconds). How often the drain monitor checks queue depth. Only used when --until-idle is active.",
)
idle_max_runtime
class-attribute
instance-attribute
¶
idle_max_runtime: float | None = Field(
default=None,
gt=0,
description="TASKQ_IDLE_MAX_RUNTIME (seconds). Maximum wall-clock time for until-idle mode. When exceeded, exit code 4. None = no limit. Only used when --until-idle is active.",
)
resolved_pg_dsn_direct
property
¶
Direct DSN guaranteed non-None after :meth:post_load.
Why a property: pg_dsn_direct: PostgresDsn | None carries the
environment-shape that distinguishes "user did not set
TASKQ_PG_DSN_DIRECT" (None, fallback to pg_dsn) from
"user set it explicitly". Once :meth:post_load has applied the
fallback, the field is always non-None - but pyright cannot
prove that across method boundaries. This property re-asserts the
invariant at every call site, eliminating the need for assert
or cast at call sites that read the DSN.
Raises :class:RuntimeError if accessed before :meth:post_load
ran (signals a programming error: WorkerSettings() constructor
must always go through :meth:load / :meth:load_from_dict).
resolved_pg_dsn_pooled
property
¶
Pooled DSN guaranteed non-None after :meth:post_load.
See :attr:resolved_pg_dsn_direct for the rationale.
worker_pool_size
property
¶
Derived pool size for worker_pool: int(max_concurrency * 1.5).
worst_case_shutdown_seconds
property
¶
Modelled worst-case wall clock from SIGTERM to process exit.
The shutdown phase graces plus the bounded-close tail that unwinds
after them (see :func:taskq._close.worst_case_teardown_tail).
This is deliberately NOT enforced by post_load. The cross-field
validator there rejects a config outright, and a budget below this
dead-backend worst case is a legitimate operator choice (a tight
dev deployment that would rather be SIGKILLed mid-unwind than wait
out a hung close). The shipped default covers the model — raising
the validator to reject sub-worst-case budgets would take that
choice away. The number is surfaced as a startup warning instead,
and docs/guides/deployment.md documents the pod-grace formula.
shutdown_budget_is_sufficient
property
¶
Whether termination_grace_period covers the modelled worst case.
post_load ¶
Apply DSN fallback and validate cross-field invariants after loading.
Runs automatically on every load path (load(),
load_from_dict(), reload(), and nested config loading),
including under validate=False - consistent with the per-field
validator hooks (transformation is part of loading, not
validation). No WorkerSettings.load / load_from_dict
override is needed; the base DotEnvConfig._load_fields invokes
this hook itself.
Returns list[ValidationError] so failures integrate with
dotenvmodel's uniform error hierarchy: a single returned error is
raised unchanged (its exact type preserved), several aggregate
into MultipleValidationErrors. Catch DotEnvModelError (the
common base) to cover both single and aggregate cases -
MultipleValidationErrors is a DotEnvModelError but not a
ValidationError, so except ValidationError alone misses the
multi-invariant case. ValidationError suffices only when at
most one invariant can fire (e.g. a single field constraint).
Source code in src/taskq/settings.py
1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 | |
FakeClock ¶
Deterministic clock for tests.
Accepts a start datetime (typically
datetime(2025, 1, 1, tzinfo=UTC)). now() returns the current
internal time; move_to and advance let tests control the clock
explicitly. monotonic() returns elapsed seconds from _EPOCH
so that elapsed-time guards see a non-zero starting value.
Source code in src/taskq/testing/clock.py
validate_actor_payload ¶
validate_actor_payload(
payload_type: type[BaseModel],
raw_payload: dict[str, object] | BaseModel,
actor: str | None = None,
) -> BaseModel
Validate a raw payload dict (or existing BaseModel) against the actor's payload model.
Wraps pydantic.ValidationError as
:class:~taskq.exceptions.PayloadValidationError (non-retryable) so
the retry classifier fails the job immediately instead of retrying
a deterministic validation failure.
Error details are sanitized via include_url=False,
include_input=False to prevent attacker-controlled field values
from being persisted to the jobs row or surfaced in the web admin.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload_type
|
type[BaseModel]
|
The actor's Pydantic payload model class. |
required |
raw_payload
|
dict[str, object] | BaseModel
|
The raw |
required |
actor
|
str | None
|
The actor name, for error context. |
None
|
Returns:
| Type | Description |
|---|---|
BaseModel
|
The validated |
Raises:
| Type | Description |
|---|---|
PayloadValidationError
|
If validation fails. |
Source code in src/taskq/_validation.py
enrich_pg_dsn ¶
Apply credential to dsn and return a self-contained DSN string.
The credential is written into the DSN userinfo (percent-encoded),
replacing any existing userinfo password - and replacing the userinfo
user when credential.username is set (Vault dynamic DB creds).
This is the only slot that is guaranteed to take effect: asyncpg's
resolver applies userinfo before query parameters (both behind
if user is None / if password is None guards), so a
query-string user= / password= is silently ignored whenever
the DSN already carries userinfo. A stale password= query
parameter is dropped (always shadowed by the userinfo password);
a user= query parameter is dropped only when the userinfo
carries a user to shadow it - a query-carried user with no userinfo
user is the effective principal and is preserved.
sslmode=require is added only when the DSN has no explicit
sslmode, so stronger modes (verify-full) are never downgraded.
Prefer the factory builders (:func:make_pg_pool_factory /
:func:make_dedicated_conn_factory) where possible - they pass the
credential as keyword arguments instead, keeping the token out of
the DSN string entirely.
Source code in src/taskq/auth.py
ensure_sslmode_require ¶
Add sslmode=require to dsn unless an sslmode is already set.
An explicit sslmode is never overridden - in particular stronger
modes (verify-ca / verify-full) must not be downgraded:
require skips certificate verification, which would expose the
very token this module injects to a MITM.
Public because anyone assembling a credential-bearing DSN by hand needs
exactly this rule and must not re-derive it: a token path that silently
connects without TLS puts the credential on the wire. The factory
builders in this module apply it for you; reach for it directly only on
the DSN paths they do not cover (a raw asyncpg.connect, a migration
connection, a DSN handed to another library).
sslmode=disable is an explicit choice and is preserved - that is how
a test container or a Unix-socket deployment opts out.
Source code in src/taskq/auth.py
make_dedicated_conn_factory ¶
make_dedicated_conn_factory(
dsn: str,
provider: PgCredentialProvider,
*,
command_timeout: float | None = None,
setup: Callable[[Connection], Awaitable[None]]
| None = None,
server_settings: dict[str, str] | None = None,
connection_class: type[Connection] | None = None,
) -> ConnFactory
Build a :data:~taskq.connections.ConnFactory backed by provider.
Used for the worker's notify_conn / leader_conn or
:class:taskq.TaskQ's pg_conn_factory. Like
:func:make_pg_pool_factory, the credential is passed as keyword
arguments (precedence over userinfo and query params; the token
never appears in the DSN string), and password= is an async
callable that asyncpg awaits per physical connection.
A dedicated connection is opened once and then held for the life of
the worker, so the callable normally fires exactly once - but these
are precisely the long-lived connections a credential expiry kills,
and the callable is what makes every re-open (a LISTEN connection
reconnecting after the server drops it, or reload_credentials
rebuilding it) authenticate with a fresh credential rather than the
one captured when the factory was first invoked.
command_timeout is forwarded to asyncpg.connect as the default
per-operation timeout. The worker's DSN-built notify_conn /
leader_conn carry dispatcher_command_timeout; pass it here too
so a credential-provider deployment does not silently drop the bound
that keeps a wedged query from stalling leader election.
setup is forwarded to asyncpg.connect and runs once after the
connection is established (e.g. registering type codecs, setting
session GUCs). For a dedicated connection this is equivalent to
init on a pool - there is no acquire/reuse cycle.
server_settings is forwarded to asyncpg.connect and applied as
session-level GUCs at connect time (e.g.
{"statement_timeout": "30s", "search_path": "app"}).
connection_class is forwarded to asyncpg.connect and sets the
:class:asyncpg.Connection subclass for this connection. Use it to
install custom codecs or override connection methods.
Source code in src/taskq/auth.py
make_pg_pool_factory ¶
make_pg_pool_factory(
dsn: str,
provider: PgCredentialProvider,
*,
min_size: int = 1,
max_size: int = 4,
max_inactive_connection_lifetime: float = 300.0,
command_timeout: float | None = None,
init: Callable[[Connection], Awaitable[None]]
| None = None,
setup: Callable[[Connection], Awaitable[None]]
| None = None,
server_settings: dict[str, str] | None = None,
connection_class: type[Connection] | None = None,
) -> PoolFactory
Build a :data:~taskq.connections.PoolFactory backed by provider.
Each invocation fetches a fresh :class:PgCredential from provider
and calls asyncpg.create_pool with the credential as keyword
arguments - password= always, user= when the credential
carries a username. Keyword arguments take precedence over both DSN
userinfo and query parameters in asyncpg's resolver, so a stale
credential baked into dsn can never shadow the fresh one, and the
token never appears in the DSN string. The pool is owned by the
worker (entered on its AsyncExitStack).
Token refresh: password= is passed as an async callable, which
asyncpg invokes and awaits once per physical connection - the
connections opened at pool creation, those opened later by pool
growth, and the replacements opened after
max_inactive_connection_lifetime recycles an idle connection. Every
new connection therefore authenticates with a freshly fetched
credential, and no external rotation is required. This matters because
Postgres authenticates at connect time only: a credential resolved once
and reused as a fixed string keeps working on already-open connections
while every new connection fails, roughly one token-lifetime after
deploy.
SIGHUP (see taskq.worker.deps.reload_credentials) still works
and is no longer required for token refresh. It remains the way to
force a full pool rebuild - and the only way to pick up a changed
username, since asyncpg resolves user= once per pool and accepts
a callable only for password=. A provider that rotates its username
(e.g. Vault dynamic database credentials) raises a RuntimeError
naming this constraint rather than pairing a fresh password with a
stale username.
Per-connection setup: init is forwarded verbatim to
asyncpg.create_pool and runs once per new physical connection
- on the connections opened at pool creation, on connections opened
later by pool growth, and again on replacements opened after
max_inactive_connection_lifetime recycles an idle connection.
That lifecycle is exactly why this setup (registering type codecs -
e.g. pgvector.asyncpg.register_vector - preparing statements,
setting session GUCs) cannot be done correctly after pool creation:
a connection configured by hand is silently replaced under load or
after an idle period. The only per-connection work this factory does
of its own is the credential refresh described above (an asyncpg
password= callback, not an init hook), so a caller-supplied
init is the only hook of its kind: it is passed through unwrapped
and can never silently replace internal setup.
Per-acquire setup: setup is forwarded to asyncpg.create_pool
and runs every time a connection is acquired from the pool
(via pool.acquire()), not just on new-connection creation. Use
it for per-checkout work that must run even when a pooled connection
is reused - e.g. resetting search_path or verifying session
state. Unlike init, setup runs on every acquire, so keep it
lightweight. Both init and setup can be provided simultaneously.
server_settings is forwarded to asyncpg.create_pool and applied
as session-level GUCs on every new connection (e.g.
{"statement_timeout": "30s", "search_path": "app"}). Useful for
per-pool configuration that must be set at connection time.
connection_class is forwarded to asyncpg.create_pool and sets
the :class:asyncpg.Connection subclass used by the pool. Use it to
install custom codecs or override connection methods across the
entire pool.
Source code in src/taskq/auth.py
306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 | |
make_redis_client_factory ¶
make_redis_client_factory(
url: str | None,
provider: RedisCredentialProvider,
**client_kwargs: Any,
) -> RedisFactory
Build a :data:~taskq.connections.RedisFactory backed by provider.
url is the Redis URL without credentials. The factory attaches
a redis-py CredentialProvider that delegates to provider, so
reconnects re-fetch the credential automatically. Use a rediss://
(TLS) URL - with a plain redis:// URL the bearer token is sent
unencrypted, and the factory logs a warning.
If url is None the factory raises :class:RuntimeError when
called (matches the worker's "Redis not configured" contract).
Source code in src/taskq/auth.py
apply_batch_terminal_outcome
async
¶
apply_batch_terminal_outcome(
backend: Backend,
job: JobRow,
outcome: AttemptOutcome,
*,
loop_conn: Connection | None = None,
) -> None
Apply batch policy after a job reaches a terminal write.
Called after every terminal write by the consumer and the in-memory
runner. For non-batched jobs (no metadata.batch_id) this returns
immediately — zero overhead.
"succeeded": resets the consecutive-failure counter. If no jobs remain non-terminal, marks the batch complete."failed": increments the consecutive-failure counter. If the threshold is reached, aborts the batch and logsbatch-aborted. If not aborted and no jobs remain non-terminal, marks the batch complete."cancelled"/"crashed": counts non-terminal jobs. If none remain, marks the batch complete."snoozed"/"reservation_denied"/"rate_limit_denied"/"scheduled": returns immediately — the job is rescheduled, not terminal.
Best-effort semantics (M7): the increment/reset/abort/complete writes are best-effort. A crash between the terminal job write and the counter increment loses that increment — the failure count is under-counted by one. The next terminal failure re-triggers the check and increments again, so a consistently failing batch still aborts (just one failure later than it would have). The stale-batch sweep is the safety net for batch STATUS (it transitions stuck active/aborted rows to terminal) but it cannot recover lost failure counts — a crash gap means the consecutive-failure streak is permanently broken, potentially preventing an abort that should have fired.
Source code in src/taskq/batch.py
306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 | |
wait_for_batch
async
¶
wait_for_batch(
db: Connection | 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
Convenience helper for the fan-out-then-finalize pattern.
Queries batch children by batch_id using the GIN-indexed
WHERE metadata @> $1::jsonb predicate.
Inside an actor (snooze_via_exception=True, the default): - If any children are in-flight, raises Snooze(snooze_interval). The consumer transitions the job to scheduled; the actor is retried after snooze_interval without consuming retry budget. - If all children are terminal, returns BatchCompletionStatus.
Outside an actor (snooze_via_exception=False): - Blocks via asyncio.sleep(snooze_interval) in a loop until all children are terminal, then returns BatchCompletionStatus. - Use this form from scripts and integration tests where no consumer is present to translate a Snooze into a rescheduled job.
expect_at_least raises :class:~taskq.exceptions.EmptyBatchError
when fewer than the expected number of jobs are present and none are
in flight. on_empty controls the behaviour when the batch has
zero jobs and no batches row exists: "error" (default) raises
:class:~taskq.exceptions.EmptyBatchError; "ok" returns the
empty status. exclude_job_id omits a specific job from the
count — when not set, the batch row's finalizer_job_id is used
automatically.
If the batch row has status = 'aborted' and all jobs are
terminal, raises :class:~taskq.exceptions.BatchAbortedError.
snooze_interval is clamped to a minimum of 1 second.
schema must match the schema used when the PostgresBackend was
constructed (default "taskq").
Source code in src/taskq/batch.py
385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 | |
register_cron ¶
Add schedule to the module-level registry.
Validates the cron expression at call time. Raises :class:ValueError
on bad expression or mutually exclusive fields. Duplicate calls append
again — deduplication is the caller's responsibility (the DB
(actor, name) UNIQUE constraint is the authoritative gate at
startup time).
Source code in src/taskq/scheduler.py
cron()¶
taskq.cron is both a submodule (src/taskq/cron.py) and, via from taskq.cron import
cron, the name of a re-exported function on the taskq package. This name collision means
the cron() function does not render under the taskq package-level directive above —
mkdocstrings resolves taskq.cron to the submodule. The explicit directive below documents
the function itself; see the Cron Scheduling guide for usage.
cron ¶
cron(
expression: str,
actor: str,
*,
payload_factory: str | None = None,
static_payload: dict[str, object] | None = None,
name: str = "",
identity_key: IdentityKey | None = None,
timezone: str = "UTC",
dst_strategy: DstStrategy = "skip",
enabled: bool = True,
) -> CronScheduleSpec
Declare a cron schedule and auto-register it.
Validates expression via croniter.is_valid(); raises
:class:ValueError on invalid expressions. Raises
:class:ValueError if both payload_factory and static_payload
are provided.
The returned :class:CronScheduleSpec is registered via
:func:~taskq.scheduler.register_cron at decoration time so
decorated schedules are auto-discovered at worker startup without
any explicit register_cron() call.
Startup auto-discovery is create-only, skip-on-conflict. Existing
cron_schedules rows are never modified by the decorator
registration pass. If a @cron decorator's parameters change
after the schedule was first registered, the operator must manually
update or delete and recreate the schedule.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dst_strategy
|
DstStrategy
|
How to handle DST gaps and overlaps.
|
'skip'
|