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

IdempotencyKey module-attribute

IdempotencyKey = NewType('IdempotencyKey', str)

Distinguishes idempotency keys from identity keys at call sites.

IdentityKey module-attribute

IdentityKey = NewType('IdentityKey', str)

Distinguishes identity keys from idempotency keys at call sites.

JobId module-attribute

JobId = NewType('JobId', UUID)

Opaque job identifier — prevents UUID mixups across the API.

QueueName module-attribute

QueueName = Annotated[
    str, AfterValidator(_validate_queue_name)
]

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__ = [
    "ActorConfigDriftError",
    "ActorConfigDriftList",
    "ActorFn",
    "ActorFnWithCtx",
    "ActorHandler",
    "ActorRef",
    "BackpressureError",
    "BatchCompletionStatus",
    "BatchHandle",
    "CancelPhase",
    "CancelResult",
    "CronScheduleSpec",
    "DIError",
    "DependencyCycle",
    "DstStrategy",
    "EnqueueItem",
    "ErrorReporter",
    "Fail",
    "IdempotencyKey",
    "IdentityKey",
    "IllegalStateTransition",
    "JobContext",
    "JobEvent",
    "JobFailed",
    "JobFilter",
    "JobHandle",
    "JobId",
    "JobRetryState",
    "JobSortField",
    "JobsClient",
    "MaxPendingExceededError",
    "MissingProvider",
    "NullErrorReporter",
    "OnSuccess",
    "PartialBatchError",
    "PayloadValidationError",
    "ProgressEvent",
    "ProgressTooLarge",
    "QueueMode",
    "QueueName",
    "RateLimitBackend",
    "ReservationUnavailable",
    "ResultTooLarge",
    "ResultUnavailable",
    "Retry",
    "RetryAfter",
    "RetryClassifier",
    "RetryClassifierHook",
    "RetryDecision",
    "RetryKind",
    "RetryOverride",
    "RetryPolicy",
    "ScheduleHandle",
    "ScheduleRecord",
    "SchemaNotMigratedError",
    "ScopeViolation",
    "SingletonCollisionError",
    "Snooze",
    "SubEnqueueError",
    "SubJobEnqueuer",
    "TaskQ",
    "TaskQError",
    "WorkerOwnershipMismatch",
    "__version__",
    "actor",
    "cron",
    "register_cron",
    "wait_for_batch",
]

__version__ module-attribute

__version__ = importlib.metadata.version('taskq-py')

ActorFn

ActorFn = Callable[[P_], Awaitable[R_]]

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

ActorFnWithCtx = Callable[
    [P_, JobContext[P_]], Awaitable[R_]
]

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.

DstStrategy

DstStrategy = Literal['skip', 'firstof', 'allof']

QueueMode

QueueMode = Literal['strict_fifo', 'round_robin']

RateLimitBackend

RateLimitBackend = Literal['redis', 'postgres', 'memory']

RetryKind

RetryKind = Literal[
    "transient", "indefinite", "non_retryable"
]

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

OnSuccess = Callable[
    [JobRow, object], Awaitable[None] | None
]

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

RetryClassifierHook = Callable[
    [BaseException, int], RetryOverride | None
]

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.

RetryDecision

RetryDecision = Retry | Fail

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.

__call__ async

__call__(
    payload: P_, /, *args: object, **kwargs: object
) -> R_

Source code in src/taskq/actor.py

async def __call__(self, payload: P_, /, *args: object, **kwargs: object) -> R_: ...

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] | None = None,
    reservations: list[str | KeyedReservationRef]
    | 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_type re-validates raw dispatch-time payloads back to P (via :meth:pydantic.BaseModel.model_validate).
• result_adapter round-trips actor returns through the JSONB result column (dump_python(mode="json") on the worker side, validate_python on the client side).

Attributes:

Name	Type	Description
max_concurrent		Fleet-wide concurrency cap for this actor. None means unbounded — matches the actor_config.max_concurrent IS NULL semantics in the dispatch CTE. Allowed values are None or int >= 0. A value of 0 means no jobs may run for this actor (useful for emergency drain scenarios). 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 actor_config.metadata (jsonb NOT NULL). Must be a plain dict[str, object] — mapping proxies and frozendicts are rejected at decoration time to avoid surprises at JSONB serialization time.

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

def __init__(
    self,
    *,
    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] | None = None,
    reservations: list[str | KeyedReservationRef] | 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,
) -> None:
    self.name = name
    self.queue = queue
    self.is_sync = is_sync
    self.wants_ctx = wants_ctx
    self.dependencies = dependencies
    self.payload_type = payload_type
    self.result_adapter = result_adapter
    self.retry = retry
    self.result_ttl = result_ttl
    self.singleton = singleton
    self.max_concurrent = max_concurrent
    self.max_pending = max_pending
    self.metadata = {} if metadata is None else metadata
    self.unique_for = unique_for
    self.unique_states = unique_states
    self.start_to_close = start_to_close
    self.rate_limits = [] if rate_limits is None else rate_limits
    self.reservations = [] if reservations is None else reservations
    self.non_retryable_exceptions = non_retryable_exceptions
    self.retry_classifier = retry_classifier
    self.on_retry_exhausted = on_retry_exhausted
    self.on_retry_exhausted_timeout = on_retry_exhausted_timeout
    self.on_success = on_success
    self.on_success_timeout = on_success_timeout
    self.priority = priority
    # Single storage slot. Call shape varies by handler — the
    # dispatcher (or :meth:`__call__`) routes based on
    # :attr:`wants_ctx`, :attr:`dependencies`, and :attr:`is_sync`.
    self._fn: Callable[..., object] = fn

__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",
)

name instance-attribute

name = name

queue instance-attribute

queue = queue

is_sync instance-attribute

is_sync = is_sync

wants_ctx instance-attribute

wants_ctx = wants_ctx

dependencies instance-attribute

dependencies = dependencies

payload_type instance-attribute

payload_type = payload_type

result_adapter instance-attribute

result_adapter = result_adapter

retry instance-attribute

retry = retry

result_ttl instance-attribute

result_ttl = result_ttl

singleton instance-attribute

singleton = singleton

max_concurrent instance-attribute

max_concurrent = max_concurrent

max_pending instance-attribute

max_pending = max_pending

metadata instance-attribute

metadata = {} if metadata is None else metadata

unique_for instance-attribute

unique_for = unique_for

unique_states instance-attribute

unique_states = unique_states

start_to_close instance-attribute

start_to_close = start_to_close

rate_limits instance-attribute

rate_limits = [] if rate_limits is None else rate_limits

reservations instance-attribute

reservations = [] if reservations is None else reservations

non_retryable_exceptions instance-attribute

non_retryable_exceptions = non_retryable_exceptions

retry_classifier instance-attribute

retry_classifier = retry_classifier

on_retry_exhausted instance-attribute

on_retry_exhausted = on_retry_exhausted

on_retry_exhausted_timeout instance-attribute

on_retry_exhausted_timeout = on_retry_exhausted_timeout

on_success instance-attribute

on_success = on_success

on_success_timeout instance-attribute

on_success_timeout = on_success_timeout

priority instance-attribute

priority = priority

fn property

fn: Callable[..., object]

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

__call__(payload: P, /, **deps: object) -> R

__call__(
    payload: P, ctx: JobContext[P], /, **deps: object
) -> R

__call__(
    payload: P,
    ctx: JobContext[P] | None = None,
    /,
    **deps: object,
) -> R

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

async def __call__(
    self,
    payload: P,
    ctx: "JobContext[P] | None" = None,
    /,
    **deps: object,
) -> R:
    """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`.
    """
    if self.is_sync:
        actor_kwargs: dict[str, object] = {"payload": payload, **deps}
        if self.wants_ctx:
            if ctx is None:
                raise TypeError(
                    f"actor {self.name!r} declares 'ctx: JobContext'; "
                    "supply a context to direct invocation"
                )
            actor_kwargs["ctx"] = ctx
        elif ctx is not None:
            raise TypeError(
                f"actor {self.name!r} does not declare a context parameter; "
                "call it with payload only"
            )
        return await asyncio.to_thread(self._fn, **actor_kwargs)  # type: ignore[return-value]  # Why: asyncio.to_thread erases the return type to Any; the caller type-narrows through ActorRef[R].
    if self.wants_ctx:
        if ctx is None:
            raise TypeError(
                f"actor {self.name!r} declares 'ctx: JobContext'; "
                "supply a context to direct invocation"
            )
        return await self._fn(payload, ctx, **deps)  # type: ignore[return-value]  # Why: _fn is typed Callable[..., object] (sync or async); caller narrows through ActorRef[R].
    if ctx is not None:
        raise TypeError(
            f"actor {self.name!r} does not declare a context parameter; "
            "call it with payload only"
        )
    return await self._fn(payload, **deps)  # type: ignore[return-value]  # Why: _fn is typed Callable[..., object] (sync or async); caller narrows through ActorRef[R].

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

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.

NONE class-attribute instance-attribute

NONE = 0

COOPERATIVE class-attribute instance-attribute

COOPERATIVE = 1

FORCED class-attribute instance-attribute

FORCED = 2

ABANDON_PENDING class-attribute instance-attribute

ABANDON_PENDING = 3

JobFilter dataclass

JobFilter(
    queue: str | None = None,
    status: 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,
)

Filter parameters for :meth:Backend.list_jobs. cursor is an opaque keyset-pagination token encoding (priority, scheduled_at, id) from the last row of the previous page. 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. See / audit M102-3.

queue class-attribute instance-attribute

queue: str | None = None

status class-attribute instance-attribute

status: JobStatus | None = None

actor class-attribute instance-attribute

actor: str | None = None

identity_key class-attribute instance-attribute

identity_key: IdentityKey | None = None

batch_id class-attribute instance-attribute

batch_id: UUID | None = None

limit class-attribute instance-attribute

limit: int = 100

cursor class-attribute instance-attribute

cursor: str | None = None

tags class-attribute instance-attribute

tags: tuple[str, ...] | None = None

order_by class-attribute instance-attribute

order_by: JobSortField | None = None

__post_init__

__post_init__() -> None

Source code in src/taskq/backend/_protocol.py

def __post_init__(self) -> None:
    if (
        self.cursor is not None
        and self.order_by is not None
        and self.order_by is not JobSortField.SCHEDULED_AT_ASC
    ):
        raise ValueError(
            "cursor pagination is only supported with the default ordering "
            "(order_by=None or JobSortField.SCHEDULED_AT_ASC); "
            "non-default order_by changes the keyset the cursor encodes"
        )

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. Cursor pagination is only valid with the default ordering; :meth:JobFilter.__post_init__ rejects a cursor combined with a non-default order_by.

SCHEDULED_AT_ASC class-attribute instance-attribute

SCHEDULED_AT_ASC = 'scheduled_at_asc'

CREATED_AT_DESC class-attribute instance-attribute

CREATED_AT_DESC = 'created_at_desc'

FINISHED_AT_DESC class-attribute instance-attribute

FINISHED_AT_DESC = 'finished_at_desc'

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.

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

id instance-attribute

id: UUID

actor instance-attribute

actor: str

name class-attribute instance-attribute

name: str = ''

cron_expr instance-attribute

cron_expr: str

timezone instance-attribute

timezone: str

dst_strategy class-attribute instance-attribute

dst_strategy: DstStrategy = 'skip'

payload_factory instance-attribute

payload_factory: str | None

identity_key class-attribute instance-attribute

identity_key: IdentityKey | None = None

enabled instance-attribute

enabled: bool

last_fired_at instance-attribute

last_fired_at: datetime | None

last_fire_error instance-attribute

last_fire_error: str | None

consecutive_failures instance-attribute

consecutive_failures: int

next_fire_at instance-attribute

next_fire_at: datetime

metadata instance-attribute

metadata: dict[str, object]

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.

total instance-attribute

total: int

pending instance-attribute

pending: int

succeeded instance-attribute

succeeded: int

failed instance-attribute

failed: int

cancelled instance-attribute

cancelled: int

crashed instance-attribute

crashed: int

abandoned instance-attribute

abandoned: int

is_complete property

is_complete: bool

True when no jobs remain in a non-terminal state.

BatchHandle

Bases: BaseModel

Handle to a group of jobs inserted by a single :meth:~taskq.client.JobsClient.enqueue_batch call.

job_handles contains one :class:~taskq.client.JobHandle per item in the original list (including idempotency-key collisions that returned existing rows). size equals len(job_handles).

:meth:status queries the database for the current completion counts of the batch.

model_config class-attribute instance-attribute

model_config = {'arbitrary_types_allowed': True}

batch_id instance-attribute

batch_id: UUID

job_handles instance-attribute

job_handles: list[Any]

List of :class:~taskq.client.JobHandle instances, one per enqueued item.

size instance-attribute

size: int

status async

status(
    db: Connection, *, schema: str = "taskq"
) -> BatchCompletionStatus

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

async def status(
    self,
    db: "asyncpg.Connection",
    *,
    schema: str = "taskq",
) -> BatchCompletionStatus:
    """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"``).
    """
    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema identifier: {schema!r}")

    containment = dumps_str({"batch_id": str(self.batch_id)})
    records = await db.fetch(
        f"SELECT status, count(*)::int AS cnt "  # noqa: S608  # Why: schema validated against _IDENT_RE immediately above.
        f'FROM "{schema}".jobs '
        "WHERE metadata @> $1::jsonb "
        "GROUP BY status",
        containment,
    )

    counts: dict[str, int] = {}
    for rec in records:
        counts[str(rec["status"])] = int(rec["cnt"])

    pending = counts.get("pending", 0) + counts.get("scheduled", 0) + counts.get("running", 0)
    return BatchCompletionStatus(
        total=sum(counts.values()),
        pending=pending,
        succeeded=counts.get("succeeded", 0),
        failed=counts.get("failed", 0),
        cancelled=counts.get("cancelled", 0),
        crashed=counts.get("crashed", 0),
        abandoned=counts.get("abandoned", 0),
    )

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.

model_config class-attribute instance-attribute

model_config = {
    "arbitrary_types_allowed": True,
    "frozen": True,
}

actor_ref instance-attribute

actor_ref: ActorRef[Any, Any]

payload instance-attribute

payload: BaseModel

scheduled_at class-attribute instance-attribute

scheduled_at: datetime | None = None

priority class-attribute instance-attribute

priority: int | None = None

fairness_key class-attribute instance-attribute

fairness_key: str | None = None

idempotency_key class-attribute instance-attribute

idempotency_key: IdempotencyKey | str | None = None

identity_key class-attribute instance-attribute

identity_key: IdentityKey | None = None

metadata class-attribute instance-attribute

metadata: dict[str, object] = Field(default_factory=dict)

tags class-attribute instance-attribute

tags: list[str] | None = None

start_to_close class-attribute instance-attribute

start_to_close: timedelta | None = None

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.

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

job_id instance-attribute

job_id: JobId

previous_status instance-attribute

previous_status: JobStatus

new_status instance-attribute

new_status: JobStatus

cancellation_initiated instance-attribute

cancellation_initiated: bool

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

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

job_id instance-attribute

job_id: JobId

status instance-attribute

status: JobStatus

progress_state instance-attribute

progress_state: dict[str, object]

progress_seq instance-attribute

progress_seq: int

terminal instance-attribute

terminal: bool

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

def __init__(
    self,
    *,
    row: JobRow,
    result_adapter: TypeAdapter[R],
    was_existing: bool,
    client: "JobsClient | None" = None,
    backend: Backend | None = None,
    _redis_client: "redis_async.Redis | None" = None,
    _settings: "TaskQSettings | None" = None,
) -> None:
    if client is None and backend is None:
        raise ValueError(
            "JobHandle requires at least one of client= or backend= (received neither)"
        )
    self._row = row
    self._result_adapter = result_adapter
    self.was_existing: bool = was_existing
    self._client = client
    self._backend: Backend = backend if backend is not None else client.backend  # pyright: ignore[reportOptionalMemberAccess]  # Why: client is guaranteed non-None when backend is None; the ValueError above ensures at least one is provided
    self._redis_client: "redis_async.Redis | None" = _redis_client  # noqa: UP037  # Why: redis_async is under TYPE_CHECKING; string annotation prevents a runtime import cycle.
    self._handle_settings: "TaskQSettings | None" = _settings  # noqa: UP037  # Why: TaskQSettings is under TYPE_CHECKING; string annotation prevents a runtime import cycle.

was_existing instance-attribute

was_existing: bool = was_existing

job_id property

job_id: JobId

The job's unique id.

actor_name property

actor_name: str

The actor this job targets.

queue property

queue: str

The queue this job was enqueued on.

status async

status() -> JobStatus

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.

Raises:

Type	Description
RuntimeError	this handle was constructed without a :class:JobsClient.

Source code in src/taskq/client/_handle.py

async def status(self) -> JobStatus:
    """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.

    Raises:
        RuntimeError: this handle was constructed without a
            :class:`JobsClient`.
    """
    if self._client is None:
        raise RuntimeError(
            "JobHandle.status() requires a JobsClient. "
            "This handle was constructed via ctx.jobs.enqueue(); use "
            "the worker's JobsClient to read job state externally."
        )
    row = await self._client.backend.get(self.job_id)
    if row is None:
        raise KeyError(self.job_id)
    return row.status

refresh async

refresh() -> JobRow

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.

Raises:

Type	Description
RuntimeError	this handle was constructed without a :class:JobsClient.

Source code in src/taskq/client/_handle.py

async def refresh(self) -> JobRow:
    """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.

    Raises:
        RuntimeError: this handle was constructed without a
            :class:`JobsClient`.
    """
    if self._client is None:
        raise RuntimeError(
            "JobHandle.refresh() requires a JobsClient. "
            "This handle was constructed via ctx.jobs.enqueue(); use "
            "the worker's JobsClient to read job state externally."
        )
    row = await self._client.backend.get(self.job_id)
    if row is None:
        raise KeyError(self.job_id)
    return row

attempts async

attempts() -> list[AttemptRow]

Return the attempt rows for this job, ordered by attempt number.

Raises:

Type	Description
RuntimeError	this handle was constructed without a :class:JobsClient.

Source code in src/taskq/client/_handle.py

async def attempts(self) -> list[AttemptRow]:
    """Return the attempt rows for this job, ordered by attempt number.

    Raises:
        RuntimeError: this handle was constructed without a
            :class:`JobsClient`.
    """
    if self._client is None:
        raise RuntimeError(
            "JobHandle.attempts() requires a JobsClient. "
            "This handle was constructed via ctx.jobs.enqueue(); use "
            "the worker's JobsClient to read job state externally."
        )
    return await self._client.backend.get_attempts(self.job_id)

cancel async

cancel(reason: str | None = None) -> CancelResult

Delegate to :meth:JobsClient.cancel.

Raises:

Type	Description
RuntimeError	this handle was constructed without a :class:JobsClient.

Source code in src/taskq/client/_handle.py

async def cancel(self, reason: str | None = None) -> CancelResult:
    """Delegate to :meth:`JobsClient.cancel`.

    Raises:
        RuntimeError: this handle was constructed without a
            :class:`JobsClient`.
    """
    if self._client is None:
        raise RuntimeError(
            "JobHandle.cancel() requires a JobsClient. "
            "This handle was constructed via ctx.jobs.enqueue(); use "
            "the worker's JobsClient to read job state externally."
        )
    return await self._client.cancel(self.job_id, reason)

wait async

wait(*, timeout: float | None = None) -> R

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.

Raises:

Type	Description
ResultUnavailable	terminal state reached but no result was stored (result TTL expired, actor returned None while R is non-None, etc.).
JobFailed	the job ended in a non-success terminal state (failed / cancelled / crashed / abandoned); the row is attached to the exception for inspection.
TimeoutError	timeout elapsed before any terminal transition was observed.

Source code in src/taskq/client/_handle.py

async def wait(self, *, timeout: float | None = None) -> R:  # noqa: ASYNC109  # Why: timeout is part of the public API contract; asyncio.timeout() context-manager doesn't fit a polling loop
    """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.

    Raises:
        ResultUnavailable: terminal state reached but no result was
            stored (result TTL expired, actor returned ``None``
            while ``R`` is non-``None``, etc.).
        JobFailed: the job ended in a non-success terminal state
            (``failed`` / ``cancelled`` / ``crashed`` / ``abandoned``);
            the row is attached to the exception for inspection.
        TimeoutError: ``timeout`` elapsed before any terminal
            transition was observed.
    """
    deadline: float | None = None
    if timeout is not None:
        deadline = asyncio.get_running_loop().time() + timeout

    while True:
        row = await self._backend.get(self.job_id)
        if row is None:
            raise KeyError(self.job_id)
        if row.status in TERMINAL_STATUSES:
            return self._extract_result(row)

        remaining: float
        if deadline is not None:
            remaining = deadline - asyncio.get_running_loop().time()
            if remaining <= 0:
                raise TimeoutError()
            sleep = min(_WAIT_POLL_INTERVAL, remaining)
        else:
            sleep = _WAIT_POLL_INTERVAL

        await asyncio.sleep(sleep)

progress_stream async

progress_stream() -> AsyncIterator[ProgressEvent]

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.

Yields events until a terminal=True event is produced.

Source code in src/taskq/client/_handle.py

async def progress_stream(self) -> AsyncIterator[ProgressEvent]:
    """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.

    Yields events until a ``terminal=True`` event is produced.
    """
    from taskq.testing.in_memory import InMemoryBackend  # lazy — test-only dep

    if isinstance(self._backend, InMemoryBackend):
        raise NotImplementedError(
            "progress_stream requires Redis; in-memory backend does not support SSE."
        )

    if self._redis_client is not None and self._handle_settings is not None:
        async for event in self._progress_stream_redis():
            yield event
    else:
        async for event in self._progress_stream_pg():
            yield event

JobsClient

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

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

def __init__(
    self,
    backend: Backend,
    *,
    clock: Clock | None = None,
    settings: "TaskQSettings | None" = None,
) -> None:
    self._backend = backend
    self._clock = clock if clock is not None else SystemClock()
    self._settings: "TaskQSettings | None" = settings  # noqa: UP037  # Why: TaskQSettings is under TYPE_CHECKING; string annotation avoids runtime import.
    self._redis_client: "redis_async.Redis | None" = None  # type: ignore[type-arg]  # noqa: UP037  # Why: redis_async is under TYPE_CHECKING; string annotation avoids runtime import. type-arg: redis-py stubs expose Redis as an unparameterised generic.
    self._exit_stack: AsyncExitStack = AsyncExitStack()
    self._warned_unique_for: set[str] = set()

backend property

backend: Backend

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

close() -> None

Close the Redis client and release resources via the exit stack.

Source code in src/taskq/client/_jobs.py

async def close(self) -> None:
    """Close the Redis client and release resources via the exit stack."""
    await self._exit_stack.aclose()
    self._redis_client = None

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,
    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 max_pending is set, a pre-flight count of pending + scheduled jobs for the actor is compared to the limit. If count >= max_pending, :class:MaxPendingExceededError is raised synchronously — the caller decides whether to retry, fail, or wait; the library does not block on capacity.

• Evaluation order at enqueue: unique_for dedup → singleton pre-flight → max_pending count check → idempotency_key INSERT → job INSERT. A unique_for hit bypasses all remaining checks; a singleton collision fires before max_pending to give the caller the more specific SingletonCollisionError.

• idempotency_key does not bypass max_pending — the idempotency ON CONFLICT fires at step 5, after the max_pending check at step 3. Re-enqueuing with a duplicate idempotency_key when the queue is full raises MaxPendingExceededError, not the deduplicated handle. Only unique_for (step 1) bypasses max_pending.

idempotency_key:

• idempotency_key is globally unique (not per-actor scoped). Callers must namespace keys to avoid collisions between actors (e.g. "actor_name:delivery_id").

• Key length is bounded at 256 characters. Empty keys and whitespace-only keys raise :class:ValueError at the client boundary before any backend call.

unique_for:

• unique_for deduplication is best-effort. Concurrent enqueues for the same (actor, identity_key) may both insert; the dispatch CTE's running_identities filter ensures only one runs.

• When either dedup mechanism matches an existing job, JobHandle.was_existing is True. This field replaces the need for callers to inspect the row's created_at to detect a dedup return.

Source code in src/taskq/client/_jobs.py

async def enqueue[P: BaseModel, R: BaseModel | None](
    self,
    ref: ActorRef[P, R],
    payload: P,
    *,
    queue: QueueName | None = None,
    scheduled_at: datetime | None = None,
    priority: int | None = None,
    schedule_to_close: datetime | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
    identity_key: IdentityKey | None = None,
    fairness_key: str | None = None,
    idempotency_key: IdempotencyKey | None = None,
    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 ``max_pending`` is set, a pre-flight count of
      ``pending`` + ``scheduled`` jobs for the actor is compared to the
      limit. If ``count >= max_pending``, :class:`MaxPendingExceededError`
      is raised synchronously — the caller decides whether to retry,
      fail, or wait; the library does not block on capacity.

    - Evaluation order at enqueue: ``unique_for`` dedup →
      singleton pre-flight → ``max_pending`` count check →
      ``idempotency_key`` INSERT → job INSERT. A ``unique_for`` hit
      bypasses all remaining checks; a singleton collision fires before
      ``max_pending`` to give the caller the more specific
      ``SingletonCollisionError``.

    - ``idempotency_key`` does **not** bypass ``max_pending`` — the
      idempotency ON CONFLICT fires at step 5, after the max_pending
      check at step 3. Re-enqueuing with a duplicate
      ``idempotency_key`` when the queue is full raises
      ``MaxPendingExceededError``, not the deduplicated handle. Only
      ``unique_for`` (step 1) bypasses max_pending.

    **idempotency_key:**

    - ``idempotency_key`` is **globally unique** (not per-actor scoped).
      Callers must namespace keys to avoid collisions between actors
      (e.g. ``"actor_name:delivery_id"``).

    - Key length is bounded at **256 characters**. Empty keys and
      whitespace-only keys raise :class:`ValueError` at the client
      boundary before any backend call.

    **unique_for:**

    - ``unique_for`` deduplication is **best-effort**. Concurrent
      enqueues for the same ``(actor, identity_key)`` may both insert;
      the dispatch CTE's ``running_identities`` filter ensures only one
      runs.

    - When either dedup mechanism matches an existing job,
      ``JobHandle.was_existing`` is ``True``. This field replaces the
      need for callers to inspect the row's ``created_at`` to detect a
      dedup return.
    """
    resolved_queue = queue if queue is not None else ref.queue
    identity_key_str = str(identity_key) if identity_key is not None else ""

    with enqueue_span(ref.name, resolved_queue, identity_key=identity_key_str) as (
        span,
        extracted_trace_id,
        extracted_span_id,
    ):
        args = build_enqueue_args(
            ref,
            payload,
            queue=queue,
            scheduled_at=scheduled_at,
            priority=priority,
            fairness_key=fairness_key,
            metadata=metadata,
            identity_key=identity_key,
            idempotency_key=idempotency_key,
            trace_id=extracted_trace_id,
            span_id=extracted_span_id,
            schedule_to_close=schedule_to_close,
            start_to_close=start_to_close,
            heartbeat_timeout=heartbeat_timeout,
            tags=tags,
            clock=self._clock,
        )
        span.set_attribute("messaging.message.id", str(args.id))
        if ref.unique_for is not None and args.identity_key is None:
            self._maybe_warn_unique_for_no_identity(ref)
        with self._translate_schema_errors():
            row = await self._backend.enqueue(args)

    if row.id == args.id:
        logger.debug(
            "job_enqueued",
            kind="job_enqueued",
            job_id=row.id,
            actor=row.actor,
            queue=row.queue,
            idempotency_key=row.idempotency_key,
        )
    return JobHandle(
        client=self,
        row=row,
        result_adapter=ref.result_adapter,
        was_existing=(row.id != args.id),
        _redis_client=self._redis_client,
        _settings=self._settings,
    )

enqueue_batch async

enqueue_batch(
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: Connection | 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.

Validation rules:

• len(items) == 0 raises :class:ValueError.
• len(items) > 1000 raises :class:ValueError.
• ALL payloads are validated before any INSERT. A single failure raises :class:~taskq.exceptions.PayloadValidationError and 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 limits 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

async def enqueue_batch(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | 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`.

    **Validation rules:**

    - ``len(items) == 0`` raises :class:`ValueError`.
    - ``len(items) > 1000`` raises :class:`ValueError`.
    - ALL payloads are validated before any INSERT.  A single failure
      raises :class:`~taskq.exceptions.PayloadValidationError` and
      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 limits
    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`).
    """
    from taskq._ids import new_job_id

    if len(items) == 0:
        raise ValueError("items must not be empty")
    if len(items) > 1000:
        raise ValueError(f"items must contain at most 1000 entries, got {len(items)}")

    # Auto-generate batch_id if not provided (UUIDv7)
    resolved_batch_id = UUID(bytes=new_job_id().bytes) if batch_id is None else batch_id

    # Phase 1: Validate ALL payloads (and idempotency keys) before any I/O
    for i, item in enumerate(items):
        ref = item.actor_ref
        try:
            ref.payload_type.model_validate(item.payload)
        except ValidationError as exc:
            # exc.errors() returns list[ErrorDetails] (TypedDict); cast to
            # the erased dict[str, object] expected by PayloadValidationError.
            errs: list[dict[str, object]] = exc.errors()  # type: ignore[assignment]  # Why: pydantic v2 ErrorDetails is a TypedDict (subtype of dict[str, Any]); assignment to list[dict[str,object]] is safe at runtime but pyright cannot prove covariance
            raise PayloadValidationError(
                f"Payload validation failed for item {i} (actor={ref.name!r}): {exc}",
                actor=ref.name,
                validation_errors=errs,
            ) from exc
        if item.idempotency_key is not None:
            if item.idempotency_key == "":
                raise ValueError(f"idempotency_key for item {i} must not be empty")
            if item.idempotency_key.strip() == "":
                raise ValueError(f"idempotency_key for item {i} must not be whitespace-only")
            if len(item.idempotency_key) > 256:
                raise ValueError(
                    f"idempotency_key for item {i} must be at most 256 characters, "
                    f"got {len(item.idempotency_key)}"
                )

    # Phase 2: Aggregated max_pending check (one query for the whole batch)
    # Collect actors that declare max_pending
    actors_with_limit: dict[str, int] = {}
    for item in items:
        ref = item.actor_ref
        mp = ref.max_pending
        if mp is not None and ref.name not in actors_with_limit:
            actors_with_limit[ref.name] = mp

    if actors_with_limit:
        # One aggregated query for all actors that declare max_pending.
        existing_counts = await self._backend.count_pending_jobs(list(actors_with_limit.keys()))
        for actor_name, limit in actors_with_limit.items():
            batch_count = sum(1 for it in items if it.actor_ref.name == actor_name)
            existing_pending_count = existing_counts.get(actor_name, 0)
            if existing_pending_count + batch_count >= limit:
                from taskq.exceptions import MaxPendingExceededError

                raise MaxPendingExceededError(
                    actor=actor_name,
                    current_count=existing_pending_count,
                    max_pending=limit,
                )

    # Phase 3: Build per-item EnqueueArgs
    args_list = build_batch_args(items, resolved_batch_id, self._clock)

    # Phase 4: Batch INSERT via backend
    rows = await self._backend.enqueue_batch(args_list, connection=connection)  # type: ignore[call-arg]  # Why: asyncpg.Connection is compatible with the protocol's connection parameter at runtime

    # Phase 5: Wrap rows in JobHandles
    handles: list[JobHandle[BaseModel | None]] = []
    for i, row in enumerate(rows):
        args = args_list[i]
        handle: JobHandle[BaseModel | None] = JobHandle(
            client=self,
            row=row,
            result_adapter=items[i].actor_ref.result_adapter,
            was_existing=(row.id != args.id),
            _redis_client=self._redis_client,
            _settings=self._settings,
        )
        handles.append(handle)

    logger.debug(
        "batch_enqueued",
        kind="batch_enqueued",
        batch_id=str(resolved_batch_id),
        size=len(items),
    )

    return BatchHandle(
        batch_id=resolved_batch_id,
        job_handles=handles,
        size=len(items),
    )

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) == 0 raises :class:ValueError.
• len(items) > 50_000 raises :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.
• 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_id to 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

async def enqueue_batch_fast(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.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) == 0`` raises :class:`ValueError`.
    - ``len(items) > 50_000`` raises :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.
    - **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_id`` to 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.
    """
    from taskq._ids import new_job_id

    if len(items) == 0:
        raise ValueError("items must not be empty")
    if len(items) > 50_000:
        raise ValueError(f"items must contain at most 50 000 entries, got {len(items)}")

    # Auto-generate batch_id if not provided (UUIDv7)
    resolved_batch_id = UUID(bytes=new_job_id().bytes) if batch_id is None else batch_id

    # Phase 1: Validate ALL payloads before any I/O
    for i, item in enumerate(items):
        ref = item.actor_ref
        try:
            ref.payload_type.model_validate(item.payload)
        except ValidationError as exc:
            errs: list[dict[str, object]] = exc.errors()  # type: ignore[assignment]  # Why: pydantic v2 ErrorDetails is a TypedDict (subtype of dict[str, Any]); assignment to list[dict[str,object]] is safe at runtime but pyright cannot prove covariance
            raise PayloadValidationError(
                f"Payload validation failed for item {i} (actor={ref.name!r}): {exc}",
                actor=ref.name,
                validation_errors=errs,
            ) from exc

    # Phase 2: Build per-item EnqueueArgs
    args_list = build_batch_args(items, resolved_batch_id, self._clock)

    # Phase 3: COPY FROM via backend
    count = await self._backend.enqueue_batch_fast(args_list, connection=connection)

    logger.debug(
        "batch_fast_enqueued",
        kind="batch_fast_enqueued",
        batch_id=str(resolved_batch_id),
        count=count,
    )

    return count

get async

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

Look up a job by id.

Returns None when the job does not exist; 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

async def get[R: BaseModel | None](
    self,
    job_id: JobId,
    *,
    result_adapter: TypeAdapter[R] | None = None,
) -> JobHandle[R] | None:
    """Look up a job by id.

    Returns ``None`` when the job does not exist; 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.
    """
    adapter: TypeAdapter[R] = (
        result_adapter if result_adapter is not None else TypeAdapter(type(None))
    )  # type: ignore[assignment]  # Why: TypeAdapter(type(None)) returns TypeAdapter[None], which does not narrow to TypeAdapter[R] under pyright; runtime behaviour is correct because None is assignable to the R bound
    with self._translate_schema_errors():
        row = await self._backend.get(job_id)
    if row is None:
        return None
    return JobHandle(
        client=self,
        row=row,
        result_adapter=adapter,
        was_existing=False,
        _redis_client=self._redis_client,
        _settings=self._settings,
    )

list async

list(filter: JobFilter) -> JobPage

List jobs matching filter, returning a :class:JobPage.

Source code in src/taskq/client/_jobs.py

async def list(self, filter: JobFilter) -> JobPage:
    """List jobs matching *filter*, returning a :class:`JobPage`."""
    with self._translate_schema_errors():
        rows = await self._backend.list_jobs(filter)
    next_cursor: str | None = None
    if rows and len(rows) == filter.limit:
        last = rows[-1]
        next_cursor = encode_cursor(last.priority, last.scheduled_at, last.id)
    return JobPage(jobs=rows, next_cursor=next_cursor)

cancel async

cancel(
    job_id: JobId, reason: str | None = None
) -> CancelResult

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

async def cancel(
    self,
    job_id: JobId,
    reason: str | None = None,
) -> CancelResult:
    """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.
    """
    from taskq.obs import record_cancel_requested

    record_cancel_requested()

    with self._translate_schema_errors():
        row = await self._backend.get(job_id)
    if row is None:
        raise KeyError(job_id)

    previous_status = row.status
    with self._translate_schema_errors():
        initiated = await self._backend.write_cancel_request(job_id, reason)
        new_row = await self._backend.get(job_id)
    if new_row is None:
        msg = (
            f"job {job_id} disappeared after write_cancel_request; "
            "the row existed a moment ago and a write was issued against it"
        )
        raise RuntimeError(msg)
    new_status = new_row.status

    result = CancelResult(
        job_id=job_id,
        previous_status=previous_status,
        new_status=new_status,
        cancellation_initiated=initiated,
    )
    logger.debug(
        "cancel_requested",
        kind="cancel_requested",
        job_id=job_id,
        previous_status=previous_status,
        cancellation_initiated=initiated,
    )
    return result

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

Parameters:

Name	Type	Description	Default
dst_strategy	DstStrategy	How to handle DST gaps and overlaps. skip (default) advances past gaps, uses the first occurrence in overlaps. firstof explicitly selects the earlier wall-clock time in overlaps. allof fires at both occurrences in overlaps.	'skip'

Source code in src/taskq/client/_jobs.py

async def create_schedule[P: BaseModel, R: BaseModel | None](
    self,
    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).

    Args:
        dst_strategy: How to handle DST gaps and overlaps.
            ``skip`` (default) advances past gaps, uses the first
            occurrence in overlaps. ``firstof`` explicitly selects the
            earlier wall-clock time in overlaps. ``allof`` fires at
            both occurrences in overlaps.
    """
    from taskq.cron import (
        ScheduleHandle,
        compute_next_fire_after,
    )

    if not croniter.is_valid(cron_expr):
        raise ValueError(f"Invalid cron expression: {cron_expr!r}")
    if payload_factory is not None and static_payload is not None:
        raise ValueError(
            "payload_factory and static_payload are mutually exclusive; "
            "provide one or the other, not both"
        )
    actor_name = actor.name if isinstance(actor, ActorRef) else actor
    # Why: actor is stored as a name string in the DB; payload type is not preserved at the cron-schedule level
    del actor

    metadata: dict[str, object] = {}
    if static_payload is not None:
        metadata["static_payload"] = static_payload

    now = datetime.now(UTC)
    next_fire = compute_next_fire_after(cron_expr, timezone, now, dst_strategy=dst_strategy)[0]

    args = ScheduleCreateArgs(
        actor=actor_name,
        cron_expr=cron_expr,
        timezone=timezone,
        next_fire_at=next_fire,
        dst_strategy=dst_strategy,
        payload_factory=payload_factory,
        enabled=enabled,
        name=name,
        identity_key=identity_key,
        metadata=metadata,
    )
    record = await self._backend.create_schedule(args)
    return ScheduleHandle(
        schedule_id=record.id,
        actor=record.actor,
        cron_expr=record.cron_expr,
        timezone=record.timezone,
        dst_strategy=record.dst_strategy,
        enabled=record.enabled,
        next_fire_at=record.next_fire_at,
        name=record.name,
        identity_key=record.identity_key,
        _backend=self._backend,
    )

list_schedules async

list_schedules(
    *, actor: str | None = None, enabled: bool | None = None
) -> list[ScheduleRecord]

List cron schedules, optionally filtered by actor or enabled status.

Source code in src/taskq/client/_jobs.py

async def list_schedules(
    self,
    *,
    actor: str | None = None,
    enabled: bool | None = None,
) -> "list[ScheduleRecord]":
    """List cron schedules, optionally filtered by actor or enabled status."""
    return await self._backend.list_schedules(actor=actor, enabled=enabled)

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

Source code in src/taskq/client/_jobs.py

async def update_schedule(
    self,
    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."
    """
    from taskq.cron import compute_next_fire_after

    if cron_expr is not None and not croniter.is_valid(cron_expr):
        raise ValueError(f"Invalid cron expression: {cron_expr!r}")
    if payload_factory is not None and static_payload is not None:
        raise ValueError(
            "payload_factory and static_payload are mutually exclusive; "
            "provide one or the other, not both"
        )

    next_fire_at: datetime | None = None
    if cron_expr is not None:
        now = datetime.now(UTC)
        records = await self._backend.list_schedules(actor=None, enabled=None)
        existing = next((r for r in records if r.id == schedule_id), None)
        tz = existing.timezone if existing is not None else "UTC"
        next_fire_at = compute_next_fire_after(cron_expr, tz, now)[0]

    metadata: dict[str, object] | None = None
    if static_payload is not None:
        metadata = {"static_payload": static_payload}

    args = ScheduleUpdateArgs(
        cron_expr=cron_expr,
        next_fire_at=next_fire_at,
        enabled=enabled,
        payload_factory=payload_factory,
        clear_payload_factory=clear_payload_factory,
        metadata=metadata,
    )
    return await self._backend.update_schedule(schedule_id, args)

delete_schedule async

delete_schedule(schedule_id: UUID) -> None

Delete a cron schedule by ID. Idempotent — no error if missing.

Source code in src/taskq/client/_jobs.py

async def delete_schedule(self, schedule_id: UUID) -> None:
    """Delete a cron schedule by ID.  Idempotent — no error if missing."""
    await self._backend.delete_schedule(schedule_id)

TaskQ

TaskQ(
    *,
    dsn: str | None = None,
    pool: Pool | 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,
    poll_timeout: float = 30.0,
)

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

def __init__(
    self,
    *,
    dsn: str | None = None,
    pool: "asyncpg.Pool | 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,
    poll_timeout: float = 30.0,
) -> None:
    if dsn is None and pool is None:
        raise ValueError("TaskQ requires either 'dsn' or 'pool'")
    if dsn is not None and pool is not None:
        raise ValueError("TaskQ accepts 'dsn' or 'pool', not both")
    if redis_url is not None and redis_client is not None:
        raise ValueError("TaskQ accepts 'redis_url' or 'redis_client', not both")

    self._dsn = dsn
    self._pool: "asyncpg.Pool | None" = pool  # noqa: UP037  # Why: asyncpg imported under TYPE_CHECKING; quotes required for runtime resolution.
    self._schema = schema
    self._min_pool_size = min_pool_size
    self._max_pool_size = max_pool_size
    self._redis_url = redis_url
    self._redis_client: "redis_async.Redis | None" = redis_client  # type: ignore[type-arg]  # noqa: UP037  # Why: erasure boundary — redis_async is under TYPE_CHECKING; string annotation avoids runtime import. type-arg: redis-py stubs expose Redis as an unparameterised generic. The caller-supplied client is stored here and forwarded to JobsClient without entering it on the exit stack.
    self._poll_timeout = poll_timeout
    self._owns_pool = pool is None
    self._client: JobsClient | None = None

open async

open() -> None

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

async def open(self) -> None:
    """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.
    """
    if self._client is not None:
        raise RuntimeError("TaskQ is already open")

    # Lazy imports keep asyncpg out of the module-level import graph so
    # taskq.testing can be imported without pulling in asyncpg.
    import asyncpg

    from taskq.backend.clock import SystemClock
    from taskq.backend.postgres import PostgresBackend
    from taskq.settings import TaskQSettings

    if self._pool is None:
        created = await asyncpg.create_pool(
            dsn=self._dsn,
            min_size=self._min_pool_size,
            max_size=self._max_pool_size,
        )
        assert created is not None  # asyncpg returns None only for record_class paths
        self._pool = created

    pool = self._pool
    assert pool is not None
    deps = _ClientDeps(
        settings=_ClientSettings(schema_name=self._schema),
        worker_pool=pool,
        heartbeat_pool=pool,
    )
    backend = PostgresBackend(
        deps,
        clock=SystemClock(),
        cancellation_grace_period=timedelta(seconds=30),
        cleanup_grace_period=timedelta(seconds=10),
    )
    settings = TaskQSettings.load_from_dict(
        {"TASKQ_SCHEMA_NAME": self._schema},
    )
    if self._redis_url is not None:
        settings.redis_url = self._redis_url  # type: ignore[assignment]  # Why: dotenvmodel PostgresDsn/RedisDsn fields accept str values at runtime but pyright cannot verify the coercion through the model's __setattr__.
    self._client = JobsClient(backend, settings=settings)
    if self._redis_client is not None:
        self._client._redis_client = self._redis_client  # pyright: ignore[reportPrivateUsage]  # Why: TaskQ owns the JobsClient lifecycle; assigning the caller-owned redis_client directly bypasses _open_redis so the client is NOT entered on the exit stack — TaskQ.close() must not close a caller-owned client.
    elif self._redis_url is not None:
        await self._client._open_redis(settings)  # pyright: ignore[reportPrivateUsage]  # Why: TaskQ owns the JobsClient lifecycle; _open_redis is the canonical hook for the owner to call after construction.

close async

close() -> None

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

async def close(self) -> None:
    """Close the client and release the pool if owned.

    Called automatically by ``__aexit__``. Safe to call explicitly.
    No-op if already closed.
    """
    if self._client is not None:
        await self._client.close()
        self._client = None
    if self._owns_pool and self._pool is not None:
        await self._pool.close()
        self._pool = None

__aenter__ async

__aenter__() -> TaskQ

Source code in src/taskq/client/_taskq.py

async def __aenter__(self) -> "TaskQ":
    await self.open()
    return self

__aexit__ async

__aexit__(*_: object) -> None

Source code in src/taskq/client/_taskq.py

async def __aexit__(self, *_: object) -> None:
    await self.close()

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

Source code in src/taskq/client/_taskq.py

async def enqueue[P: BaseModel, R: BaseModel | None](
    self,
    ref: ActorRef[P, R],
    payload: P,
    *,
    queue: QueueName | None = None,
    scheduled_at: datetime | None = None,
    priority: int | None = None,
    schedule_to_close: datetime | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
    identity_key: IdentityKey | None = None,
    fairness_key: str | None = None,
    idempotency_key: IdempotencyKey | None = None,
    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."""
    return await self._require_open().enqueue(
        ref,
        payload,
        queue=queue,
        scheduled_at=scheduled_at,
        priority=priority,
        schedule_to_close=schedule_to_close,
        start_to_close=start_to_close,
        heartbeat_timeout=heartbeat_timeout,
        identity_key=identity_key,
        fairness_key=fairness_key,
        idempotency_key=idempotency_key,
        trace_id=trace_id,
        span_id=span_id,
        metadata=metadata,
        tags=tags,
    )

enqueue_batch async

enqueue_batch(
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: Connection | 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

async def enqueue_batch(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | 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.
    """
    return await self._require_open().enqueue_batch(
        items,
        batch_id=batch_id,
        connection=connection,
    )

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

async def enqueue_batch_fast(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.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`.
    """
    return await self._require_open().enqueue_batch_fast(
        items,
        batch_id=batch_id,
        connection=connection,
    )

get async

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

Look up a job by id. Returns None when the job does not exist.

Source code in src/taskq/client/_taskq.py

async def get[R: BaseModel | None](
    self,
    job_id: JobId,
    *,
    result_adapter: TypeAdapter[R] | None = None,
) -> JobHandle[R] | None:
    """Look up a job by id. Returns ``None`` when the job does not exist."""
    return await self._require_open().get(job_id, result_adapter=result_adapter)

list async

list(filter: JobFilter) -> JobPage

List jobs matching filter, returning a :class:JobPage.

Source code in src/taskq/client/_taskq.py

async def list(self, filter: JobFilter) -> JobPage:
    """List jobs matching *filter*, returning a :class:`JobPage`."""
    return await self._require_open().list(filter)

cancel async

cancel(
    job_id: JobId, reason: str | None = None
) -> CancelResult

Request cancellation of a job. Raises :class:KeyError if not found.

Source code in src/taskq/client/_taskq.py

async def cancel(
    self,
    job_id: JobId,
    reason: str | None = None,
) -> CancelResult:
    """Request cancellation of a job. Raises :class:`KeyError` if not found."""
    return await self._require_open().cancel(job_id, reason)

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

async def create_schedule[P: BaseModel, R: BaseModel | None](
    self,
    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.
    """
    return await self._require_open().create_schedule(
        actor,
        cron_expr,
        timezone=timezone,
        dst_strategy=dst_strategy,
        payload_factory=payload_factory,
        static_payload=static_payload,
        name=name,
        identity_key=identity_key,
        enabled=enabled,
    )

list_schedules async

list_schedules(
    *, actor: str | None = None, enabled: bool | None = None
) -> list[ScheduleRecord]

List cron schedules. Delegates to :meth:JobsClient.list_schedules.

Source code in src/taskq/client/_taskq.py

async def list_schedules(
    self,
    *,
    actor: str | None = None,
    enabled: bool | None = None,
) -> "list[ScheduleRecord]":
    """List cron schedules.  Delegates to :meth:`JobsClient.list_schedules`."""
    return await self._require_open().list_schedules(actor=actor, enabled=enabled)

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

async def update_schedule(
    self,
    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`."""
    return await self._require_open().update_schedule(
        schedule_id,
        cron_expr=cron_expr,
        enabled=enabled,
        payload_factory=payload_factory,
        static_payload=static_payload,
        clear_payload_factory=clear_payload_factory,
    )

delete_schedule async

delete_schedule(schedule_id: UUID) -> None

Delete a cron schedule. Delegates to :meth:JobsClient.delete_schedule.

Source code in src/taskq/client/_taskq.py

async def delete_schedule(self, schedule_id: UUID) -> None:
    """Delete a cron schedule.  Delegates to :meth:`JobsClient.delete_schedule`."""
    await self._require_open().delete_schedule(schedule_id)

stream async

stream(job_id: JobId) -> AsyncIterator[JobEvent]

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

async def stream(self, job_id: JobId) -> AsyncIterator[JobEvent]:
    """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()}\n\n"

    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).
    """
    client = self._require_open()
    row = await client.backend.get(job_id)
    if row is None:
        raise KeyError(job_id)

    event = _row_to_event(row)
    yield event
    if event.terminal:
        return

    if self._redis_client is not None:
        async for evt in _stream_redis(
            self._redis_client,
            self._schema,
            job_id,
            client,
            self._poll_timeout,
            last_seq=row.progress_seq,
            last_status=row.status,
        ):
            yield evt
            if evt.terminal:
                return
    else:
        async for evt in _stream_pg(
            self._dsn,
            self._schema,
            job_id,
            client,
            self._poll_timeout,
            last_seq=row.progress_seq,
            last_status=row.status,
        ):
            yield evt
            if evt.terminal:
                return

SubJobEnqueuer

SubJobEnqueuer(
    loop_scope_resolved: Mapping[type, object] | None,
    worker_pool: Pool | None,
    backend: Backend,
    *,
    clock: Clock | 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

def __init__(
    self,
    loop_scope_resolved: Mapping[type, object] | None,
    worker_pool: asyncpg.Pool | None,
    backend: Backend,
    *,
    clock: Clock | None = None,
) -> None:
    self._loop_scope_resolved = loop_scope_resolved
    self._worker_pool = worker_pool
    self._backend = backend
    self._clock = clock if clock is not None else SystemClock()
    self._pending_buffer: list[EnqueueArgs] = []
    self._loop_enqueue_args: list[EnqueueArgs] = []
    self._autonomous_enqueue_count: int = 0

pending_count property

pending_count: int

pending_items property

pending_items: Sequence[EnqueueArgs]

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,
    unique_for: timedelta | None = None,
    unique_states: tuple[JobStatus, ...] | None = None,
    max_pending: int | None = None,
) -> JobHandle[R]

Source code in src/taskq/client/_enqueuer.py

async def enqueue[P: BaseModel, R: BaseModel | None](
    self,
    actor_ref: ActorRef[P, R],
    payload: P,
    *,
    connection: asyncpg.Connection | None = None,
    scheduled_at: datetime | None = None,
    priority: int | None = None,
    fairness_key: str | None = None,
    metadata: dict[str, object] | None = None,
    identity_key: IdentityKey | None = None,
    idempotency_key: IdempotencyKey | str | None = None,
    unique_for: timedelta | None = None,
    unique_states: tuple[JobStatus, ...] | None = None,
    max_pending: int | None = None,
) -> JobHandle[R]:
    resolved_queue = actor_ref.queue
    identity_key_str = str(identity_key) if identity_key is not None else ""

    with enqueue_span(actor_ref.name, resolved_queue, identity_key=identity_key_str) as (
        span,
        extracted_trace_id,
        extracted_span_id,
    ):
        args = build_enqueue_args(
            actor_ref,
            payload,
            scheduled_at=scheduled_at,
            priority=priority,
            fairness_key=fairness_key,
            metadata=metadata,
            identity_key=identity_key,
            idempotency_key=idempotency_key,
            trace_id=extracted_trace_id,
            span_id=extracted_span_id,
            unique_for=unique_for,
            unique_states=unique_states,
            max_pending=max_pending,
            clock=self._clock,
        )
        span.set_attribute("messaging.message.id", str(args.id))
        row = await self._do_enqueue(args, connection)
    return JobHandle(
        row=row,
        result_adapter=actor_ref.result_adapter,
        was_existing=(row.id != args.id),
        backend=self._backend,
        client=None,
    )

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

Source code in src/taskq/client/_enqueuer.py

async def enqueue_batch(
    self,
    items: Sequence[EnqueueItem[Any, Any]],
    *,
    batch_id: UUID | None = None,
    connection: asyncpg.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).
    """
    resolved_batch_id = batch_id if batch_id is not None else UUID(bytes=new_job_id().bytes)

    conn, from_loop_scope = self._resolve_connection(connection)

    if conn is not None:
        args_list = build_batch_args(items, resolved_batch_id, self._clock)

        if from_loop_scope and self._backend.supports_transactional_simulation:
            for args in args_list:
                self._pending_buffer.append(args)
            return [
                JobHandle(
                    row=self._synthesize_row(args),
                    result_adapter=item.actor_ref.result_adapter,
                    was_existing=False,
                    backend=self._backend,
                    client=None,
                )
                for args, item in zip(args_list, items, strict=True)
            ]

        rows = await self._backend.enqueue_batch(args_list, connection=conn)  # type: ignore[call-arg]  # Why: asyncpg.Connection is compatible with the protocol's connection parameter at runtime
        if from_loop_scope:
            self._loop_enqueue_args.extend(args_list)
        handles: list[JobHandle[Any]] = []
        for i, row in enumerate(rows):
            args = args_list[i]
            handles.append(
                JobHandle(
                    row=row,
                    result_adapter=items[i].actor_ref.result_adapter,
                    was_existing=(row.id != args.id),
                    backend=self._backend,
                    client=None,
                )
            )
        return handles

    if self._worker_pool is None:
        raise RuntimeError("ctx.jobs is only available inside an actor body")

    handles = []
    failed_items: list[tuple[int, Exception]] = []
    for i, item in enumerate(items):
        item_metadata: dict[str, object] = dict(item.metadata)
        item_metadata["batch_id"] = str(resolved_batch_id)
        try:
            handle = await self.enqueue(
                item.actor_ref,
                item.payload,
                scheduled_at=item.scheduled_at,
                priority=item.priority,
                fairness_key=item.fairness_key,
                metadata=item_metadata,
                idempotency_key=item.idempotency_key,
                identity_key=item.identity_key,
            )
            handles.append(handle)
        except Exception as exc:
            failed_items.append((i, exc))

    if failed_items:
        raise PartialBatchError(
            succeeded_count=len(handles),
            failed_items=failed_items,
            total=len(items),
        )

    return handles

flush_buffer async

flush_buffer() -> None

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

async def flush_buffer(self) -> None:
    """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.
    """
    snapshot = self._pending_buffer
    self._pending_buffer = []
    self._loop_enqueue_args.clear()
    failed_items: list[tuple[EnqueueArgs, Exception]] = []
    for args in snapshot:
        try:
            await self._backend.enqueue(args)
        except Exception as exc:
            failed_items.append((args, exc))
            _log.warning(
                "sub_enqueue_flush_error",
                kind="sub_enqueue_flush_error",
                job_id=args.id,
                message=str(exc),
            )
    if failed_items:
        raise SubEnqueueError(failed_items=failed_items)

discard_buffer

discard_buffer() -> None

Clear the pending buffer without flushing.

Source code in src/taskq/client/_enqueuer.py

def discard_buffer(self) -> None:
    """Clear the pending buffer without flushing."""
    self._pending_buffer.clear()
    self._loop_enqueue_args.clear()

drain_for_re_enqueue

drain_for_re_enqueue() -> list[EnqueueArgs]

Return and clear both loop-scope and pending buffers for re-enqueue.

Source code in src/taskq/client/_enqueuer.py

def drain_for_re_enqueue(self) -> list[EnqueueArgs]:
    """Return and clear both loop-scope and pending buffers for re-enqueue."""
    items = self._loop_enqueue_args + list(self._pending_buffer)
    self._loop_enqueue_args = []
    self._pending_buffer = []
    return items

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.

job_id instance-attribute

job_id: UUID

actor instance-attribute

actor: str

queue instance-attribute

queue: str

attempt instance-attribute

attempt: int

worker_id instance-attribute

worker_id: UUID

payload instance-attribute

payload: P

jobs instance-attribute

jobs: SubJobEnqueuer

log instance-attribute

log: BoundLogger

span class-attribute instance-attribute

span: Span | None = None

cancel_event class-attribute instance-attribute

cancel_event: Event = field(default_factory=asyncio.Event)

cancellation_requested property

cancellation_requested: bool

check_cancelled

check_cancelled() -> None

Source code in src/taskq/context.py

def check_cancelled(self) -> None:
    if self.cancel_event.is_set():
        raise asyncio.CancelledError

should_abort

should_abort() -> bool

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	True when cancellation has been requested — the sync
bool	actor should return or raise immediately.

Source code in src/taskq/context.py

def should_abort(self) -> bool:
    """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:
        ``True`` when cancellation has been requested — the sync
        actor should return or raise immediately.
    """
    return self._abort_requested.is_set()

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

async def progress(
    self,
    *,
    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.
    """
    if data is not None and self._worker_settings is not None:
        serialised_len = len(dumps(data))
        limit = self._worker_settings.progress_data_max_bytes
        if serialised_len > limit:
            raise ProgressTooLarge(limit=limit, actual=serialised_len)

    if self._progress_buffers is None:
        return

    buffer = self._progress_buffers.get(self.job_id)
    if buffer is None:
        return

    buffer.pending_seq_delta += 1
    if step is not None:
        buffer.pending_state["step"] = step
    if percent is not None:
        buffer.pending_state["percent"] = percent
    if detail is not None:
        buffer.pending_state["detail"] = detail
    if data is not None:
        buffer.pending_state["data"] = data
    buffer.dirty = True

    seq = buffer.base_seq + buffer.pending_seq_delta

    if self._redis_client is not None and self._worker_settings is not None:
        coro = _publish_progress_event(
            self._redis_client,
            self._worker_settings,
            self.actor,
            self.job_id,
            step=step,
            percent=percent,
            detail=detail,
            data=data,
            seq=seq,
        )
        if self._pending_publish_tasks is not None:
            task = asyncio.create_task(coro, name=f"taskq-progress-publish-{self.job_id}")
            self._pending_publish_tasks.add(task)
            task.add_done_callback(self._pending_publish_tasks.discard)
        else:
            # No shared task set to hold a reference (e.g. a caller
            # constructing JobContext directly without a WorkerDeps) —
            # fall back to awaiting inline rather than risking the
            # scheduled task being garbage-collected mid-publish.
            await coro

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.

actor instance-attribute

actor: str

cron_expr instance-attribute

cron_expr: str

timezone class-attribute instance-attribute

timezone: str = 'UTC'

dst_strategy class-attribute instance-attribute

dst_strategy: DstStrategy = 'skip'

payload_factory class-attribute instance-attribute

payload_factory: str | None = None

static_payload class-attribute instance-attribute

static_payload: dict[str, object] | None = None

name class-attribute instance-attribute

name: str = ''

identity_key class-attribute instance-attribute

identity_key: IdentityKey | None = None

enabled class-attribute instance-attribute

enabled: bool = True

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.

schedule_id instance-attribute

schedule_id: UUID

actor instance-attribute

actor: str

cron_expr instance-attribute

cron_expr: str

timezone instance-attribute

timezone: str

enabled instance-attribute

enabled: bool

next_fire_at instance-attribute

next_fire_at: datetime

dst_strategy class-attribute instance-attribute

dst_strategy: DstStrategy = 'skip'

name class-attribute instance-attribute

name: str = ''

identity_key class-attribute instance-attribute

identity_key: IdentityKey | None = None

disable async

disable() -> None

Source code in src/taskq/cron.py

async def disable(self) -> None:
    await self._backend.update_schedule(
        self.schedule_id,
        ScheduleUpdateArgs(enabled=False),
    )

enable async

enable() -> None

Source code in src/taskq/cron.py

async def enable(self) -> None:
    await self._backend.update_schedule(
        self.schedule_id,
        ScheduleUpdateArgs(enabled=True),
    )

delete async

delete() -> None

Source code in src/taskq/cron.py

async def delete(self) -> None:
    await self._backend.delete_schedule(self.schedule_id)

ActorConfigDriftError

ActorConfigDriftError(
    actor: str,
    field: Literal[
        "max_concurrent",
        "max_pending",
        "queue",
        "result_ttl",
        "metadata",
    ],
    registered: int
    | float
    | str
    | dict[str, object]
    | None,
    stored: int | float | str | dict[str, object] | None,
)

Bases: TaskQError

One actor whose registered config differs from the stored row.

Source code in src/taskq/exceptions.py

def __init__(
    self,
    actor: str,
    field: Literal["max_concurrent", "max_pending", "queue", "result_ttl", "metadata"],
    registered: int | float | str | dict[str, object] | None,
    stored: int | float | str | dict[str, object] | None,
) -> None:
    self.actor = actor
    self.field = field
    self.registered = registered
    self.stored = stored
    super().__init__(
        f"ActorConfigDrift: actor={actor}, field={field}, "
        f"registered={registered!r}, stored={stored!r}"
    )

hint class-attribute instance-attribute

hint = _ACTOR_CONFIG_DRIFT_HINT

actor instance-attribute

actor = actor

field instance-attribute

field = field

registered instance-attribute

registered = registered

stored instance-attribute

stored = stored

ActorConfigDriftList

ActorConfigDriftList(
    drifts: tuple[ActorConfigDriftError, ...],
)

Bases: TaskQError

Collected wrapper raised at worker startup when one or more actors have drift.

Source code in src/taskq/exceptions.py

def __init__(self, drifts: tuple[ActorConfigDriftError, ...]) -> None:
    self.drifts = drifts
    lines = [f"{len(drifts)} actor(s) have config drift:"]
    for d in drifts:
        lines.append(f"  {d}")
    lines.append(self.hint)
    super().__init__("\n".join(lines))

hint class-attribute instance-attribute

hint = _ACTOR_CONFIG_DRIFT_HINT

drifts instance-attribute

drifts = drifts

BackpressureError

BackpressureError(
    actor: str,
    pending: int = 0,
    max_pending: int | None = None,
)

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

def __init__(self, actor: str, pending: int = 0, max_pending: int | None = None) -> None:
    self.actor = actor
    self.pending = pending
    self.max_pending = max_pending
    super().__init__(
        f"BackpressureError: actor={actor}, pending={pending}, max_pending={max_pending}"
    )

actor instance-attribute

actor = actor

pending instance-attribute

pending = pending

max_pending instance-attribute

max_pending = max_pending

DependencyCycle

DependencyCycle(cycle_path: list[str])

Bases: TaskQError

A cycle was detected in the provider graph.

Source code in src/taskq/exceptions.py

def __init__(self, cycle_path: list[str]) -> None:
    if len(cycle_path) < 2:
        raise ValueError(
            f"cycle_path must contain at least 2 entries (got {len(cycle_path)!r})"
        )
    self.cycle_path = list(cycle_path)
    super().__init__(f"dependency cycle: {' -> '.join(cycle_path)}")

cycle_path instance-attribute

cycle_path = list(cycle_path)

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.

IllegalStateTransition

IllegalStateTransition(
    job_id: JobId,
    from_status: JobStatus,
    to_status: JobStatus,
)

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

def __init__(
    self,
    job_id: JobId,
    from_status: JobStatus,
    to_status: JobStatus,
) -> None:
    self.job_id = job_id
    self.from_status = from_status
    self.to_status = to_status
    super().__init__(
        f"job {self.job_id} cannot transition from {self.from_status} to {self.to_status}"
    )

job_id instance-attribute

job_id = job_id

from_status instance-attribute

from_status = from_status

to_status instance-attribute

to_status = to_status

JobFailed

JobFailed(row: JobRow)

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

def __init__(self, row: "JobRow") -> None:
    self.row = row
    super().__init__(
        f"job {row.id} ended in {row.status!r}"
        + (f": {row.error_class}: {row.error_message}" if row.error_class else ""),
    )

row instance-attribute

row = row

MaxPendingExceededError

MaxPendingExceededError(
    actor: str, current_count: int, max_pending: int
)

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

def __init__(self, actor: str, current_count: int, max_pending: int) -> None:
    self.current_count = current_count
    super().__init__(actor, pending=current_count, max_pending=max_pending)

current_count instance-attribute

current_count = current_count

MissingProvider

MissingProvider(*, type_name: str, required_by: str)

Bases: TaskQError

A type was injected but no provider is registered.

Source code in src/taskq/exceptions.py

def __init__(self, *, type_name: str, required_by: str) -> None:
    self.type_name = type_name
    self.required_by = required_by
    super().__init__(f"no provider registered for {type_name} (required by {required_by})")

type_name instance-attribute

type_name = type_name

required_by instance-attribute

required_by = required_by

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

def __init__(
    self,
    *,
    succeeded_count: int,
    failed_items: list[tuple[int, Exception]],
    total: int,
) -> None:
    self.succeeded_count = succeeded_count
    self.failed_items = failed_items
    self.total = total
    super().__init__(
        f"PartialBatchError: {succeeded_count}/{total} succeeded, "
        f"{len(failed_items)} failed at indices: {[i for i, _ in failed_items]}"
    )

succeeded_count instance-attribute

succeeded_count = succeeded_count

failed_items instance-attribute

failed_items = failed_items

total instance-attribute

total = total

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

def __init__(
    self,
    detail: str,
    *,
    actor: str | None = None,
    payload_schema_ver: str | None = None,
    validation_errors: list[dict[str, object]] | None = None,
) -> None:
    self.actor = actor
    self.payload_schema_ver = payload_schema_ver
    self.validation_errors: list[dict[str, object]] = validation_errors or []
    super().__init__(detail)

actor instance-attribute

actor = actor

payload_schema_ver instance-attribute

payload_schema_ver = payload_schema_ver

validation_errors instance-attribute

validation_errors: list[dict[str, object]] = (
    validation_errors or []
)

ProgressTooLarge

ProgressTooLarge(limit: int, actual: int)

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

def __init__(self, limit: int, actual: int) -> None:
    self.limit = limit
    self.actual = actual
    super().__init__(f"Progress data payload {actual}B exceeds limit {limit}B")

limit instance-attribute

limit = limit

actual instance-attribute

actual = actual

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

def __init__(
    self,
    bucket_name: str,
    retry_after: timedelta,
    *,
    source: Literal["reservation", "rate_limit"] = "reservation",
) -> None:
    if retry_after < timedelta(0):
        raise ValueError(f"retry_after must be non-negative, got {retry_after!r}")
    super().__init__(f"no reservation slot in {bucket_name!r}")
    self.bucket_name = bucket_name
    self.retry_after = retry_after
    self.source = source

bucket_name instance-attribute

bucket_name = bucket_name

retry_after instance-attribute

retry_after = retry_after

source instance-attribute

source = source

ResultTooLarge

Bases: TaskQError

Terminal result exceeded the 64KB cap.

ResultUnavailable

ResultUnavailable(row: JobRow)

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

def __init__(self, row: "JobRow") -> None:
    self.row = row
    super().__init__(f"job {row.id} has no stored result")

row instance-attribute

row = row

RetryAfter

RetryAfter(
    delay: timedelta, *, consume_budget: bool = True
)

Bases: TaskQError

Schedule retry at specific delay. Consumes retry budget by default.

Source code in src/taskq/exceptions.py

def __init__(self, delay: timedelta, *, consume_budget: bool = True) -> None:
    if delay < timedelta(0):
        raise ValueError(f"delay must be non-negative, got {delay!r}")
    super().__init__(f"retry after {delay}")
    self.delay = delay
    self.consume_budget = consume_budget

delay instance-attribute

delay = delay

consume_budget instance-attribute

consume_budget = consume_budget

SchemaNotMigratedError

SchemaNotMigratedError(schema: str)

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

def __init__(self, schema: str) -> None:
    self.schema = schema
    super().__init__(
        f"TaskQ schema {schema!r} is missing or not migrated. "  # noqa: S608  # Why: human-readable error message, not a SQL query; ruff's SQL-injection heuristic false-positives on the word "schema" near f-string interpolation.
        f"Run `taskq migrate up` to create/update it, or set "
        f"TASKQ_MIGRATE_ON_START=true to migrate automatically at worker startup."
    )

schema instance-attribute

schema = schema

ScopeViolation

ScopeViolation(
    *,
    from_scope: Scope,
    to_scope: Scope,
    type_name: str,
    dependent: str,
)

Bases: TaskQError

A provider depends on a shorter-lived scope than its own.

Source code in src/taskq/exceptions.py

def __init__(
    self,
    *,
    from_scope: Scope,
    to_scope: Scope,
    type_name: str,
    dependent: str,
) -> None:
    self.from_scope = from_scope
    self.to_scope = to_scope
    self.type_name = type_name
    self.dependent = dependent
    super().__init__(
        f"{from_scope.name}-scoped {dependent} depends on {to_scope.name}-scoped {type_name}"
    )

from_scope instance-attribute

from_scope = from_scope

to_scope instance-attribute

to_scope = to_scope

type_name instance-attribute

type_name = type_name

dependent instance-attribute

dependent = dependent

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

def __init__(
    self,
    actor: str,
    blocking_job_id: UUID | None = None,
    retry_after: timedelta | None = None,
) -> None:
    self.blocking_job_id = blocking_job_id
    self.retry_after = retry_after
    super().__init__(actor)

blocking_job_id instance-attribute

blocking_job_id = blocking_job_id

retry_after instance-attribute

retry_after = retry_after

Snooze

Snooze(delay: timedelta)

Bases: TaskQError

Job returns control with new scheduled_at; does not consume retry budget.

Source code in src/taskq/exceptions.py

def __init__(self, delay: timedelta) -> None:
    if delay < timedelta(0):
        raise ValueError(f"delay must be non-negative, got {delay!r}")
    super().__init__(f"snooze for {delay}")
    self.delay = delay

delay instance-attribute

delay = delay

SubEnqueueError

SubEnqueueError(
    failed_items: list[tuple[EnqueueArgs, Exception]],
)

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

def __init__(
    self,
    failed_items: "list[tuple[EnqueueArgs, Exception]]",
) -> None:
    self.failed_items = failed_items
    super().__init__(
        f"SubEnqueueError: {len(failed_items)} sub-job(s) failed to enqueue after parent commit"
    )

failed_items instance-attribute

failed_items = failed_items

TaskQError

Bases: Exception

Base for all library-raised exceptions.

WorkerOwnershipMismatch

WorkerOwnershipMismatch(
    job_id: UUID, expected: UUID, actual: UUID | None
)

Bases: TaskQError

Terminal write predicate failed: job exists but is owned by a different worker.

Source code in src/taskq/exceptions.py

def __init__(
    self,
    job_id: UUID,
    expected: UUID,
    actual: UUID | None,
) -> None:
    self.job_id = job_id
    self.expected = expected
    self.actual = actual
    super().__init__(f"job {self.job_id} owned by {actual}, expected {expected}")

job_id instance-attribute

job_id = job_id

expected instance-attribute

expected = expected

actual instance-attribute

actual = actual

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.

report async

report(job: JobRow, exception: BaseException) -> None

Source code in src/taskq/obs/error_reporter.py

async def report(self, job: JobRow, exception: BaseException) -> None: ...

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.

report async

report(job: JobRow, exception: BaseException) -> None

Source code in src/taskq/obs/error_reporter.py

async def report(self, job: JobRow, exception: BaseException) -> None:
    return None

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.

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

v class-attribute instance-attribute

v: int = 1

kind instance-attribute

kind: Literal['progress', 'state_change']

job_id instance-attribute

job_id: UUID

actor instance-attribute

actor: str

ts instance-attribute

ts: datetime

seq instance-attribute

seq: int

status instance-attribute

status: str

step class-attribute instance-attribute

step: int | None = None

percent class-attribute instance-attribute

percent: float | None = None

detail class-attribute instance-attribute

detail: str | None = None

data class-attribute instance-attribute

data: dict[str, object] | None = None

terminal class-attribute instance-attribute

terminal: bool = False

Fail

Bases: BaseModel

Fail decision: the job will not be retried.

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

error_class instance-attribute

error_class: str

retryable instance-attribute

retryable: bool

JobRetryState

Bases: NamedTuple

Projection of JobRow columns consumed by the retry classifier.

attempt instance-attribute

attempt: int

max_attempts instance-attribute

max_attempts: int

retry_kind instance-attribute

retry_kind: RetryKind

schedule_to_close instance-attribute

schedule_to_close: datetime | None

start_to_close instance-attribute

start_to_close: timedelta | None

Retry

Bases: BaseModel

Retry decision: reschedule the job at next_scheduled_at.

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

next_scheduled_at instance-attribute

next_scheduled_at: datetime

RetryClassifier

Pure classifier that maps an exception + policy to a RetryDecision.

classify staticmethod

classify(
    policy: RetryPolicy,
    non_retryable_exceptions: tuple[
        type[BaseException], ...
    ],
    exception: BaseException,
    attempt: int,
    schedule_to_close: datetime | None,
    now: datetime,
    *,
    max_retry_backoff: timedelta = timedelta(hours=24),
    override: RetryOverride | None = None,
) -> RetryDecision

Source code in src/taskq/retry.py

@staticmethod
def classify(
    policy: RetryPolicy,
    non_retryable_exceptions: tuple[type[BaseException], ...],
    exception: BaseException,
    attempt: int,
    schedule_to_close: datetime | None,
    now: datetime,
    *,
    max_retry_backoff: timedelta = timedelta(hours=24),
    override: RetryOverride | None = None,
) -> RetryDecision:
    if isinstance(exception, non_retryable_exceptions):
        return Fail(error_class=type(exception).__name__, retryable=False)

    if isinstance(exception, PayloadValidationError):
        return Fail(error_class="PayloadValidationError", retryable=False)

    effective_kind = (
        override.kind if override is not None and override.kind is not None else policy.kind
    )
    override_delay = override.delay if override is not None else None

    if effective_kind == "non_retryable":
        return Fail(error_class=type(exception).__name__, retryable=False)

    if effective_kind == "transient":
        if attempt < policy.max_attempts:
            return RetryClassifier._retry_or_deadline(
                policy,
                attempt,
                schedule_to_close,
                now,
                max_retry_backoff=max_retry_backoff,
                override_delay=override_delay,
            )
        return Fail(error_class=type(exception).__name__, retryable=False)

    # effective_kind == "indefinite"
    if schedule_to_close is not None and now >= schedule_to_close:
        return Fail(error_class="DeadlineExceeded", retryable=False)
    return RetryClassifier._retry_or_deadline(
        policy,
        attempt,
        schedule_to_close,
        now,
        max_retry_backoff=max_retry_backoff,
        override_delay=override_delay,
    )

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.

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

kind class-attribute instance-attribute

kind: RetryKind | None = None

delay class-attribute instance-attribute

delay: timedelta | None = None

RetryPolicy

Bases: BaseModel

Policy controlling retry behaviour for an actor.

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

kind class-attribute instance-attribute

kind: RetryKind = 'transient'

max_attempts class-attribute instance-attribute

max_attempts: int = 3

time_budget class-attribute instance-attribute

time_budget: timedelta | None = None

backoff class-attribute instance-attribute

backoff: Literal["exponential", "linear", "fixed"] = (
    "exponential"
)

base class-attribute instance-attribute

base: timedelta = timedelta(seconds=5)

cap class-attribute instance-attribute

cap: timedelta = timedelta(hours=1)

jitter class-attribute instance-attribute

jitter: float = 0.2

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

Empty batch: if batch_id matches no rows, returns BatchCompletionStatus(total=0, pending=0, is_complete=True) and logs a WARNING (may indicate a wrong batch_id).

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

async def wait_for_batch(
    db: "asyncpg.Connection | asyncpg.Pool",
    batch_id: UUID,
    *,
    schema: str = "taskq",
    snooze_interval: timedelta = timedelta(seconds=10),
    snooze_via_exception: bool = True,
) -> 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.

    Empty batch: if batch_id matches no rows, returns
    BatchCompletionStatus(total=0, pending=0, is_complete=True) and logs
    a WARNING (may indicate a wrong batch_id).

    snooze_interval is clamped to a minimum of 1 second.
    ``schema`` must match the schema used when the PostgresBackend was
    constructed (default ``"taskq"``).
    """
    from taskq.exceptions import Snooze

    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema identifier: {schema!r}")

    _min_snooze = timedelta(seconds=1)

    import asyncpg as _asyncpg

    if snooze_interval < _min_snooze:
        original = snooze_interval
        snooze_interval = _min_snooze
        _logger.warning(
            "snooze-interval-clamped",
            original=str(original),
            clamped=str(snooze_interval),
        )

    sql = _WAIT_FOR_BATCH_SQL.format(schema=schema)
    containment = dumps_str({"batch_id": str(batch_id)})

    async def _query(conn: "asyncpg.Connection") -> BatchCompletionStatus:
        row = await conn.fetchrow(sql, containment)
        if row is None:
            return BatchCompletionStatus(
                total=0,
                pending=0,
                succeeded=0,
                failed=0,
                cancelled=0,
                crashed=0,
                abandoned=0,
            )
        status = BatchCompletionStatus(
            total=int(row["total"]),
            pending=int(row["in_flight"]),
            succeeded=int(row["succeeded"]),
            failed=int(row["failed"]),
            cancelled=int(row["cancelled"]),
            crashed=int(row["crashed"]),
            abandoned=int(row["abandoned"]),
        )
        if status.total == 0:
            _logger.warning(
                "wait-for-batch-empty",
                batch_id=str(batch_id),
            )
        return status

    if isinstance(db, _asyncpg.Pool):
        async with db.acquire() as conn:  # type: ignore[reportArgumentType]  # Why: Pool.acquire() returns PoolConnectionProxy; pyright stubs model it as incompatible with Connection but it is runtime-compatible
            status = await _query(conn)  # type: ignore[reportArgumentType]  # Why: PoolConnectionProxy is a runtime-compatible Connection proxy; pyright stubs model it as incompatible
    else:
        status = await _query(db)

    if status.pending > 0:
        if snooze_via_exception:
            raise Snooze(snooze_interval)
        while status.pending > 0:
            await asyncio.sleep(snooze_interval.total_seconds())
            if isinstance(db, _asyncpg.Pool):
                async with db.acquire() as conn:  # type: ignore[reportArgumentType]  # Why: Pool.acquire() returns PoolConnectionProxy; pyright stubs model it as incompatible with Connection but it is runtime-compatible
                    status = await _query(conn)  # type: ignore[reportArgumentType]  # Why: PoolConnectionProxy is a runtime-compatible Connection proxy; pyright stubs model it as incompatible
            else:
                status = await _query(db)

    return status

register_cron

register_cron(schedule: CronScheduleSpec) -> None

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

def register_cron(schedule: CronScheduleSpec) -> None:
    """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).
    """
    _validate_spec(schedule)
    _CRON_REGISTRY.append(schedule)

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 (default) advances past gaps, uses the first occurrence in overlaps. firstof explicitly selects the earlier wall-clock time in overlaps. allof fires at both occurrences in overlaps (the caller receives two datetimes from compute_next_fire_after).	'skip'

Source code in src/taskq/cron.py

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

    Args:
        dst_strategy: How to handle DST gaps and overlaps.
            ``skip`` (default) advances past gaps, uses the first
            occurrence in overlaps. ``firstof`` explicitly selects the
            earlier wall-clock time in overlaps. ``allof`` fires at
            both occurrences in overlaps (the caller receives two
            datetimes from ``compute_next_fire_after``).
    """
    if not croniter.is_valid(expression):
        raise ValueError(f"Invalid cron expression: {expression!r}")
    if payload_factory is not None and static_payload is not None:
        raise ValueError(
            "payload_factory and static_payload are mutually exclusive; "
            "provide one or the other, not both"
        )
    spec = CronScheduleSpec(
        actor=actor,
        cron_expr=expression,
        timezone=timezone,
        dst_strategy=dst_strategy,
        payload_factory=payload_factory,
        static_payload=static_payload,
        name=name,
        identity_key=identity_key,
        enabled=enabled,
    )
    from taskq.scheduler import register_cron

    register_cron(spec)
    return spec