Client¶

TaskQ, JobsClient, JobHandle, CancelResult, and ActorsClient.

_taskq ¶

Top-level entry point for non-worker applications.

Provides :class:TaskQ, a Postgres-backed client that manages its own connection pool and exposes job operations (enqueue, get, list, cancel) directly.

Two lifecycle patterns are supported:

Async context manager (scripts, tests)::

async with TaskQ(dsn="postgresql://user:pw@host/db") as tq:
    handle = await tq.enqueue(my_actor, MyPayload(...))
    result = await handle.wait()

Explicit open/close (FastAPI lifespan, long-lived processes)::

tq = TaskQ(dsn=settings.pg_dsn)

@asynccontextmanager
async def lifespan(app: FastAPI):
    await tq.open()
    yield
    await tq.close()

@app.post("/tasks")
async def create_task(payload: MyPayload):
    handle = await tq.enqueue(my_actor, payload)
    return {"job_id": str(handle.job_id)}

Passing an existing pool (e.g. shared with the rest of the application)::

async with TaskQ(pool=app.state.pool) as tq:
    await tq.cancel(job_id)

all `module-attribute` ¶

__all__ = [
    "ActorsClient",
    "EventRow",
    "JobEvent",
    "TaskQ",
    "orjson_response_class",
]

logger `module-attribute` ¶

logger = structlog.get_logger('taskq.client._taskq')

EventRow `dataclass` ¶

EventRow(
    event_id: int,
    job_id: JobId,
    occurred_at: datetime,
    kind: Literal["state_change", "cancel_request"],
    detail: dict[str, object],
)

Read-model of a taskq.job_events row.

Mirrors the job_events table shape: monotonic event_id, the owning job, timestamp, event kind, and a detail payload.

event_id `instance-attribute` ¶

event_id: int

job_id `instance-attribute` ¶

job_id: JobId

occurred_at `instance-attribute` ¶

occurred_at: datetime

kind `instance-attribute` ¶

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

detail `instance-attribute` ¶

detail: dict[str, object]

ActorsClient ¶

ActorsClient(pool: Pool, *, schema: str = 'taskq')

Pool-wrapping facade for actor configuration operations.

Acquires a connection from the injected pool for each call, delegates to taskq.actor_config_ops, and returns the result. The caller must have opened the pool; this class does not manage its lifecycle.

.. note:: This client is Postgres-only, it delegates to :mod:taskq.actor_config_ops, which executes raw SQL against the actor_config table. The :class:~taskq.backend._protocol.Backend protocol does not include actor config operations, so :class:~taskq.testing.InMemoryBackend does not support ActorsClient.

Parameters¶

pool: An open asyncpg.Pool. The caller retains ownership. schema: TaskQ schema name. Defaults to "taskq".

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

def __init__(self, pool: "asyncpg.Pool", *, schema: str = "taskq") -> None:
    self._pool = pool
    self._schema = schema

list `async` ¶

list() -> list[ActorConfigRow]

List all stored actor_config rows, ordered by actor name.

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

async def list(self) -> list[ActorConfigRow]:
    """List all stored actor_config rows, ordered by actor name."""
    async with self._pool.acquire() as conn:
        return await list_actor_configs(conn, schema=self._schema)

get `async` ¶

get(actor: str) -> ActorConfigRow | None

Get one actor_config row, or None if not found.

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

async def get(self, actor: str) -> ActorConfigRow | None:
    """Get one actor_config row, or ``None`` if not found."""
    async with self._pool.acquire() as conn:
        return await get_actor_config(conn, actor, schema=self._schema)

set_capacity `async` ¶

set_capacity(
    actor: str,
    *,
    max_concurrent: int | Unset | None = UNSET,
    max_pending: int | Unset | None = UNSET,
    result_ttl: float | Unset | None = UNSET,
) -> ActorConfigRow | None

Update capacity fields on an existing actor_config row.

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

async def set_capacity(
    self,
    actor: str,
    *,
    max_concurrent: int | Unset | None = UNSET,
    max_pending: int | Unset | None = UNSET,
    result_ttl: float | Unset | None = UNSET,
) -> ActorConfigRow | None:
    """Update capacity fields on an existing actor_config row."""
    async with self._pool.acquire() as conn:
        return await set_actor_config_capacity(
            conn,
            actor,
            max_concurrent=max_concurrent,
            max_pending=max_pending,
            result_ttl=result_ttl,
            schema=self._schema,
        )

deregister `async` ¶

deregister(
    actor: str,
    *,
    force: bool = False,
    purge_queue: bool = False,
) -> DeregisterResult

Deregister an actor with safety checks.

See :func:taskq.actor_config_ops.deregister_actor for the full semantics.

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

async def deregister(
    self,
    actor: str,
    *,
    force: bool = False,
    purge_queue: bool = False,
) -> DeregisterResult:
    """Deregister an actor with safety checks.

    See :func:`taskq.actor_config_ops.deregister_actor` for
    the full semantics.
    """
    async with self._pool.acquire() as conn:
        return await deregister_actor(
            conn,
            actor,
            force=force,
            purge_queue=purge_queue,
            schema=self._schema,
        )

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. Both transports below route through orjson (:func:taskq._json.dumps), never stdlib json::

# HTTP: drop-in replacement for starlette's stdlib-json JSONResponse
OrjsonJSONResponse = orjson_response_class()

@app.get("/tasks/{job_id}/events")
async def events(job_id: UUID) -> Response:
    event = latest_event(job_id)
    return OrjsonJSONResponse(event.model_dump())

# WebSocket: send_json uses stdlib json.dumps internally, send the
# pre-rendered orjson text instead:
async for event in tq.stream(job_id):
    await websocket.send_text(event.model_dump_json())

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

TaskQ ¶

TaskQ(
    *,
    dsn: str | None = None,
    pool: Pool | None = None,
    pool_factory: PoolFactory | None = None,
    pg_provider: PgCredentialProvider | None = None,
    schema: str | None = None,
    min_pool_size: int = 1,
    max_pool_size: int = 5,
    redis_url: str | None = None,
    redis_client: Any | None = None,
    pg_conn_factory: ConnFactory | None = None,
    listen_conn: Connection | None = None,
    poll_timeout: float = 30.0,
    reclaim_event_visibility_delay: timedelta | None = None,
    reload_interval: float | None = None,
)

Postgres-backed TaskQ client.

Manages a connection pool and exposes job operations directly. Supports both the async context manager pattern and explicit open() / close() for frameworks like FastAPI that manage their own lifecycle.

Parameters¶

dsn: Postgres DSN string. Mutually exclusive with pool. pool: An already-open asyncpg.Pool. The caller retains ownership; close() will not close it. pool_factory: A zero-arg async factory returning an asyncpg.Pool - the same :data:~taskq.connections.PoolFactory the worker takes via :class:~taskq.connections.WorkerConnections. TaskQ invokes it at open() and owns the result (close() closes it), and :meth:reload_credentials re-invokes it to rotate the pool in place. This is the client-side equivalent of a <role>_pool_factory: pair it with :func:taskq.auth.make_pg_pool_factory for a rotating-credential deployment. Mutually exclusive with dsn and pool. pg_provider: A :class:~taskq.auth.PgCredentialProvider. Sugar for pool_factory=make_pg_pool_factory(dsn, pg_provider, min_size=min_pool_size, max_size=max_pool_size) - it is exactly that call and nothing more, so the two paths share one mechanism. Requires dsn; use pool_factory directly when you need the factory's other hooks (init, server_settings, command_timeout). schema: TaskQ schema name. When omitted, resolved the way the worker and CLI resolve it, TASKQ_SCHEMA_NAME from the process environment or the .env cascade, with "taskq" as the final default, so a client constructed with no opinion lands in the schema its worker fleet is listening on. An explicit value always wins. min_pool_size: Minimum pool connections. Only used when dsn or pg_provider is provided (both build the pool through the same factory path; a caller-supplied pool / pool_factory is used as-is). max_pool_size: Maximum pool connections. Only used when dsn or pg_provider is provided. redis_url: Redis URL string. Mutually exclusive with redis_client. The library creates and owns the Redis client; close() will close it. redis_client: An already-open redis.asyncio.Redis client. The caller retains ownership; close() will not close it. Mutually exclusive with redis_url. pg_conn_factory: A zero-arg async factory returning an asyncpg.Connection for the LISTEN/NOTIFY transport used by :meth:watch_reclaims. Mutually exclusive with listen_conn. Takes precedence over dsn when set. Use this when you have no DSN (e.g. AAD-managed-identity auth) but still want the reclaim wake-up. TaskQ owns and closes the connection produced by the factory per watch_reclaims() call. listen_conn: A pre-constructed asyncpg.Connection for the LISTEN transport. Caller-owned; TaskQ does not close it. Mutually exclusive with pg_conn_factory. Takes precedence over dsn when set. Use this to share a dedicated LISTEN conn across callers. poll_timeout: Maximum seconds to wait between transport wakeups before re-fetching job state. Defaults to 30.0. :meth:stream on Postgres alone has no wakeups and polls every min(poll_timeout, 0.5) seconds (never below 0.1 s, jittered ±20%). reload_interval: Seconds between automatic :meth:reload_credentials rebuilds of a factory-built pool (pool_factory / pg_provider); the client-side TASKQ_RELOAD_INTERVAL. When None (the default) the cadence is derived from the lease the provider grants - a Vault dynamic credential is rebuilt at half its lease TTL (see :class:taskq.auth.ReloadSchedule) - and a factory whose credential carries no lease is never rebuilt automatically (token providers refresh per connection and need no rebuild). The rotation runs as a background task for the life of the client; a rebuild that fails leaves the live pool serving and is retried on the next tick. Rejected for dsn= and pool= (a ValueError at construction), which have nothing to rebuild from — it is not silently ignored, the docstring's earlier claim. reclaim_event_visibility_delay: Overrides the margin :meth:watch_reclaims assumes between a job_events writer's INSERT and its COMMIT (see :data:taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY, default 2s). It is passed through to the constructed backend's poll_reclaim_events and bounds the NOTIFY catch-up retry. Must match whatever margin the worker fleet's sweep-adjacent backend uses: the margin's correctness depends on writer transaction duration, not reader preference, so a client that widens it alone only silences the gap detection on itself. None (the default) uses the shipped constant.

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

def __init__(
    self,
    *,
    dsn: str | None = None,
    pool: "asyncpg.Pool | None" = None,
    pool_factory: "PoolFactory | None" = None,
    pg_provider: "PgCredentialProvider | None" = None,
    schema: str | None = None,
    min_pool_size: int = 1,
    max_pool_size: int = 5,
    redis_url: str | None = None,
    redis_client: Any | None = None,
    pg_conn_factory: "ConnFactory | None" = None,
    listen_conn: "asyncpg.Connection | None" = None,
    poll_timeout: float = 30.0,
    reclaim_event_visibility_delay: timedelta | None = None,
    reload_interval: float | None = None,
) -> None:
    # True when every connection the client will use came from a pool
    # TaskQ built itself, so its per-query bound is known here. A
    # caller-supplied pool or factory is caller-owned, its timeouts
    # are its choice, the same doctrine the worker applies.
    self._pool_bound_is_ours = pool is None and pool_factory is None
    if pg_provider is not None:
        if dsn is None:
            raise ValueError(
                "TaskQ 'pg_provider' requires 'dsn', the provider issues a credential, "
                "not a host. Pass 'pool_factory' instead when there is no DSN."
            )
        if pool_factory is not None:
            raise ValueError("TaskQ accepts 'pg_provider' or 'pool_factory', not both")
        # Why here and not in open(): the factory is the single mechanism
        # (taskq.auth.make_pg_pool_factory, the same builder
        # build_worker_connections uses); pg_provider is sugar that
        # collapses into it, so everything downstream sees one code path.
        from taskq.auth import ReloadSchedule, make_pg_pool_factory

        pool_factory = make_pg_pool_factory(
            dsn,
            pg_provider,
            min_size=min_pool_size,
            max_size=max_pool_size,
            # Derived from the operator's lock budgets, not the bare
            # floor: the budgets are env-configured, so the derivation
            # reads the same values open()'s settings load will see.
            command_timeout=_client_pool_command_timeout_secs(),
            reload_schedule=ReloadSchedule(configured=reload_interval),
        )
        dsn = None

    sources = [
        name
        for name, v in (("dsn", dsn), ("pool", pool), ("pool_factory", pool_factory))
        if v is not None
    ]
    if not sources:
        raise ValueError("TaskQ requires one of 'dsn', 'pool' or 'pool_factory'")
    if len(sources) > 1:
        raise ValueError(f"TaskQ accepts one of {sources!r}, 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")
    if redis_url is not None and not redis_url.strip():
        # Why: dotenvmodel coerces "" to None for the optional field, so
        # the os.getenv(..., "") anti-pattern would pass the is-not-None
        # guard and silently disable Redis instead of failing at startup.
        raise ValueError("TaskQ 'redis_url' must be a non-empty URL or None")
    if pg_conn_factory is not None and listen_conn is not None:
        raise ValueError("TaskQ accepts 'pg_conn_factory' or 'listen_conn', not both")

    self._dsn = dsn
    self._pool: "asyncpg.Pool | None" = pool  # noqa: UP037  # Why: asyncpg imported under TYPE_CHECKING; quotes required for runtime resolution.
    # Schema resolution keeps one source of truth with the worker and
    # CLI (the ``ui_serve`` idiom): an explicit argument wins, otherwise
    # the shared configuration cascade decides: process environment,
    # then the .env cascade, then the model default, the same layers
    # WorkerSettings.load() reads (see _resolve_default_schema_name:
    # the value is validated ALONE, so an unrelated malformed TASKQ_*
    # var cannot break an embedder's constructor). Hardcoding
    # the default here splits that truth: TASKQ_SCHEMA_NAME honored by
    # the worker fleet but not by the client is a silent job-loss
    # vector (the enqueue succeeds into a schema no worker reads;
    # wait() reports only a bare timeout).
    if schema is None:
        schema = _resolve_default_schema_name()
    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._pg_conn_factory = pg_conn_factory
    self._listen_conn = listen_conn
    self._poll_timeout = poll_timeout
    # Passed through to the constructed PostgresBackend's poll_reclaim_events
    # default, see RECLAIM_EVENT_VISIBILITY_DELAY. Must match whatever
    # margin the worker fleet's sweep-adjacent backend uses, since the
    # margin's correctness depends on writer transaction duration, not
    # reader preference.
    self._reclaim_event_visibility_delay = reclaim_event_visibility_delay
    self._pool_factory = pool_factory
    # A pool TaskQ built (from a DSN or a factory) is TaskQ's to close; a
    # caller-supplied one never is. Same ownership rule as
    # taskq.connections.WorkerConnections.
    self._owns_pool = pool is None
    self._client: JobsClient | None = None
    self._actors_client: ActorsClient | None = None
    # Held so reload_credentials can swap the pool into the live backend
    # without reaching through the JobsClient into PostgresBackend._deps.
    self._deps: _ClientDeps | None = None
    if reload_interval is not None and reload_interval <= 0:
        raise ValueError(
            f"TaskQ 'reload_interval' must be a positive number of seconds, "
            f"got {reload_interval!r}"
        )
    if reload_interval is not None and pool_factory is None:
        raise ValueError(
            "TaskQ 'reload_interval' requires 'pool_factory' (or 'pg_provider'): a "
            "pool built from 'dsn' or passed as 'pool=' has no factory to rebuild from"
        )
    self._reload_schedule = self._make_reload_schedule(pool_factory, reload_interval)
    self._reload_task: asyncio.Task[None] | None = None
    # Serializes reload_credentials: the scheduled rotation and an
    # explicit call racing each other would each build a pool and one
    # would close the other's fresh one.
    self._reload_lock = asyncio.Lock()

actors `property` ¶

actors: ActorsClient

Actor configuration client, list, get, set capacity, deregister.

Raises RuntimeError if called before open() or outside an async with block.

backend `property` ¶

backend: Backend

The Postgres :class:Backend this client built, read-only.

The seam a host that embeds the admin UI needs: create_router takes the backend verbatim (backend=tq.backend) so the admin portal's backend-mediated mutations (job cancel, job retry, schedule run-now) serve from the SAME backend the trigger app enqueues through, instead of the 503 "not configured" the router answers when it has none. Raises RuntimeError if called before open() or outside an async with block, like the other accessors.

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

    # The fork guard pins THIS process as the resource owner before any
    # socket exists: the preload-in-master-then-fork embedding shape
    # (gunicorn --preload, a multiprocessing fork context) leaves every
    # child holding this pool's sockets, and the guard is what turns a
    # child's use of them from silent wire corruption into a typed
    # refusal. Idempotent; a no-op on platforms without fork.
    install_fork_guard()

    # 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

    # Route the Redis URL through load_from_dict so it is coerced and
    # validated by the field's declared RedisDsn type (TypeCoercionError
    # on an invalid scheme) instead of being stored as a raw str. Loaded
    # before pool creation so the DSN-built pool resolves its
    # statement-cache kwargs through the same settings instance the
    # client hands the backend, one settings flow, not two (and
    # validation failure now fails fast, before a pool is opened). The
    # lock-budget overlay folds the operator's TASKQ_*_LOCK_TIMEOUT_MS
    # values, resolved through the SAME env/.env cascade the worker's
    # load() reads (see _lock_budget_env_overlay), into the dict load:
    # that loader reads only the dict, and the constructor's own
    # arguments stay authoritative (they are written last).
    load_data: dict[str, str] = _lock_budget_env_overlay()
    load_data["TASKQ_SCHEMA_NAME"] = self._schema
    if self._redis_url is not None:
        load_data["TASKQ_REDIS_URL"] = self._redis_url
    settings = TaskQSettings.load_from_dict(load_data)

    # The per-query bound the enqueue lock budgets must fit inside,
    # derived BEFORE the pool is built: at the shipped defaults it is
    # exactly _CLIENT_POOL_COMMAND_TIMEOUT_SECS (the floor), and a
    # budget widened past its default re-derives the bound upward so
    # the widened server-side lock_timeout still fires first, the
    # clamp below then never silently caps an operator's widening.
    # Known only for pools TaskQ builds itself (the DSN pool below and
    # the pg_provider sugar's factory, which __init__ derived the same
    # way); a caller-supplied pool or factory is caller-owned, its
    # timeouts are its choice, and guessing one would clamp a budget
    # against a bound that is not there.
    pool_command_timeout = (
        lock_budget_command_timeout_secs(
            _lock_budget_pairs(settings), floor_secs=_CLIENT_POOL_COMMAND_TIMEOUT_SECS
        )
        if self._pool_bound_is_ours
        else None
    )

    if self._pool is None:
        if self._pool_factory is not None:
            # Why bounded: this open runs before anything else exists
            # , no pool, no watchdog, so an unbounded factory call
            # (the AAD first token fetch lives inside it) wedges the
            # client process forever with no signal.
            # _RELOAD_FACTORY_TIMEOUT_SECS is the SAME bound the
            # worker applies to its bootstrap factory calls
            # (worker/deps.py), not a second mechanism.
            try:
                self._pool = await asyncio.wait_for(
                    self._pool_factory(), timeout=_RELOAD_FACTORY_TIMEOUT_SECS
                )
            except TimeoutError as exc:
                raise TimeoutError(
                    f"TaskQ.open(): pool_factory did not return within "
                    f"{_RELOAD_FACTORY_TIMEOUT_SECS}s, the credential "
                    "provider behind it (e.g. a token endpoint) is "
                    "black-holed. The open fails loudly; nothing was "
                    "built."
                ) from exc
        else:
            stmt_kwargs = statement_cache_kwargs(settings)
            created = await asyncpg.create_pool(
                dsn=self._dsn,
                min_size=self._min_pool_size,
                max_size=self._max_pool_size,
                command_timeout=pool_command_timeout,
                statement_cache_size=stmt_kwargs["statement_cache_size"],
                max_cached_statement_lifetime=stmt_kwargs["max_cached_statement_lifetime"],
                connection_class=guarded_connection_class(),
            )
            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,
            # Clamped against the pool's own per-query bound: a budget
            # the connection will not wait out cannot produce the typed
            # refusal it exists for (see bounded_lock_budget_ms). The
            # bound above was derived from these same budgets, so the
            # clamp is the safety net that keeps the server-side
            # lock_timeout first, not a silent cap on a widened budget.
            # Pools the caller supplies keep their own timeouts, so
            # their budgets stand as configured.
            max_pending_lock_timeout_ms=bounded_lock_budget_ms(
                settings.max_pending_lock_timeout_ms, pool_command_timeout
            ),
            unique_for_lock_timeout_ms=bounded_lock_budget_ms(
                settings.unique_for_lock_timeout_ms, pool_command_timeout
            ),
            idempotency_lock_timeout_ms=bounded_lock_budget_ms(
                settings.idempotency_lock_timeout_ms, pool_command_timeout
            ),
        ),
        worker_pool=pool,
        heartbeat_pool=pool,
    )
    self._deps = deps
    backend = PostgresBackend(
        deps,
        clock=SystemClock(),
        cancellation_grace_period=timedelta(seconds=30),
        cleanup_grace_period=timedelta(seconds=10),
        reclaim_event_visibility_delay=(
            self._reclaim_event_visibility_delay
            if self._reclaim_event_visibility_delay is not None
            else RECLAIM_EVENT_VISIBILITY_DELAY
        ),
    )
    self._client = JobsClient(backend, settings=settings)
    self._actors_client = ActorsClient(pool, schema=self._schema)
    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.
    self._start_credential_rotation()

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._reload_task is not None:
        self._reload_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._reload_task
        self._reload_task = None
    if self._client is not None:
        await self._client.close()
        self._client = None
        self._actors_client = None
        self._deps = None
    if self._owns_pool and self._pool is not None:
        # Why bounded: an enqueue in flight at close time can stall
        # Pool.close() indefinitely against a dead PG.
        await close_pool_bounded(self._pool, "client", CLOSE_TIMEOUT_SECS)
        self._pool = None

reload_credentials `async` ¶

reload_credentials() -> None

Rebuild the pool from pool_factory and swap it in, live.

The client-side counterpart of :func:taskq.worker.deps.reload_credentials, and the supported replacement for reaching into _pool / _client: it invokes the factory (which fetches a fresh credential), atomically points every subsystem that holds a pool - the backend deps behind enqueue/get/list/cancel, the :attr:actors client, and the stream() poll - at the new pool, then closes the old one with the same bounded drain used at close().

A username-bearing pair (a Vault lease) is rotated automatically on the client's :class:~taskq.auth.ReloadSchedule - reload_interval when given, otherwise half the granted lease TTL - so this is the on-demand path: call it on any signal your issuer gives you, or to drop sessions opened under a revoked token. It is not needed for the ordinary token refresh that :func:taskq.auth.make_pg_pool_factory already does per physical connection. Concurrent calls (the scheduled rotation and an explicit one) are serialized.

Raises :class:RuntimeError if the client is not open, or if the pool is caller-owned (pool=) - TaskQ must never close a pool it does not own, so the caller rotates that one itself.

If the factory fails, the exception propagates and the current pool is left untouched and serving: a transient token-endpoint outage must not turn into a client outage.

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

async def reload_credentials(self) -> None:
    """Rebuild the pool from ``pool_factory`` and swap it in, live.

    The client-side counterpart of
    :func:`taskq.worker.deps.reload_credentials`, and the supported
    replacement for reaching into ``_pool`` / ``_client``: it invokes the
    factory (which fetches a fresh credential), atomically points every
    subsystem that holds a pool - the backend deps behind
    ``enqueue``/``get``/``list``/``cancel``, the :attr:`actors` client, and
    the ``stream()`` poll - at the new pool, then closes the old
    one with the same bounded drain used at ``close()``.

    A username-bearing pair (a Vault lease) is rotated automatically on
    the client's :class:`~taskq.auth.ReloadSchedule` - ``reload_interval``
    when given, otherwise half the granted lease TTL - so this is the
    on-demand path: call it on any signal your issuer gives you, or to
    drop sessions opened under a revoked token. It is not needed for
    the ordinary token refresh that :func:`taskq.auth.make_pg_pool_factory`
    already does per physical connection. Concurrent calls (the
    scheduled rotation and an explicit one) are serialized.

    Raises :class:`RuntimeError` if the client is not open, or if the pool
    is caller-owned (``pool=``) - TaskQ must never close a pool it does not
    own, so the caller rotates that one itself.

    If the factory fails, the exception propagates and the **current pool
    is left untouched and serving**: a transient token-endpoint outage
    must not turn into a client outage.
    """
    if self._client is None or self._deps is None:
        raise RuntimeError("TaskQ is not open, call open() before reload_credentials()")
    if self._pool_factory is None:
        raise RuntimeError(
            "TaskQ.reload_credentials() requires 'pool_factory' (or 'pg_provider'); "
            "a pool passed as 'pool=' is caller-owned and must be rotated by its owner."
        )

    async with self._reload_lock:
        old_pool = self._pool
        # Built before anything is swapped, so a factory failure leaves the
        # live pool in place, see the docstring. Why bounded: a hung token
        # endpoint must fail the rotation loudly (the live pool keeps
        # serving), never wedge the client, the SAME bound the worker's
        # reload_credentials applies to its identical factory calls.
        try:
            new_pool = await asyncio.wait_for(
                self._pool_factory(), timeout=_RELOAD_FACTORY_TIMEOUT_SECS
            )
        except TimeoutError as exc:
            raise TimeoutError(
                f"TaskQ.reload_credentials(): pool_factory did not return "
                f"within {_RELOAD_FACTORY_TIMEOUT_SECS}s, the live pool is "
                "untouched and still serving. Check the credential provider "
                "behind TaskQ 'pool_factory'."
            ) from exc

        self._pool = new_pool
        self._deps.worker_pool = new_pool
        self._deps.heartbeat_pool = new_pool
        # Rebuilt rather than mutated: ActorsClient takes its pool at
        # construction and is a cheap, stateless facade over it.
        self._actors_client = ActorsClient(new_pool, schema=self._schema)

        if old_pool is not None:
            # Why bounded: an enqueue in flight on the old pool can stall
            # Pool.close() indefinitely against a dead PG.
            await close_pool_bounded(old_pool, "client-reload", CLOSE_TIMEOUT_SECS)

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,
    idempotency_scope: str | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    metadata: dict[str, object] | None = None,
    tags: list[str] | None = None,
) -> JobHandle[R]

Enqueue a job and return a typed handle.

The actor's name is NOT validated against the registration table at enqueue time (any ActorRef is accepted, the same defer-to-fire-time doctrine create_schedule applies): a ref whose actor no worker declares enqueues successfully and the job is parked at the snooze cadence by every worker that claims it (released_reason: "actor-not-found", the stranded-jobs detector surfaces it) rather than being refused here. See :meth:JobsClient.enqueue for the full enqueue contract: dedup, max_pending, and idempotency semantics.

schedule_to_close (absolute datetime) is deprecated, it crosses clock domains (the app clock that produced it vs the database clock that evaluates it). Declare retry.time_budget on the actor instead; the interval form is anchored to the database clock.

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

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,
    idempotency_scope: str | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    metadata: dict[str, object] | None = None,
    tags: list[str] | None = None,
) -> JobHandle[R]:
    """Enqueue a job and return a typed handle.

    The actor's name is NOT validated against the registration table
    at enqueue time (any ``ActorRef`` is accepted, the same
    defer-to-fire-time doctrine ``create_schedule`` applies): a
    ref whose actor no worker declares enqueues successfully and the
    job is parked at the snooze cadence by every worker that claims
    it (``released_reason: "actor-not-found"``, the stranded-jobs
    detector surfaces it) rather than being refused here. See
    :meth:`JobsClient.enqueue` for the full enqueue contract:
    dedup, ``max_pending``, and idempotency semantics.

    ``schedule_to_close`` (absolute datetime) is deprecated, it crosses
    clock domains (the app clock that produced it vs the database clock
    that evaluates it).  Declare ``retry.time_budget`` on the actor
    instead; the interval form is anchored to the database clock.
    """
    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,
        idempotency_scope=idempotency_scope,
        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,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | None = None,
) -> BatchHandle

Enqueue multiple jobs in a single batched INSERT.

Delegates to :meth:JobsClient.enqueue_batch; see its docstring for validation rules and idempotency-key collision semantics.

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

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

enqueue_batch_streaming `async` ¶

enqueue_batch_streaming(
    items: Iterable[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: Connection | None = None,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | None = None,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> BatchHandle

Enqueue jobs from a lazy iterable in chunks.

Delegates to :meth:JobsClient.enqueue_batch_streaming; see its docstring for chunk_size validation and streaming semantics.

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

async def enqueue_batch_streaming(
    self,
    items: Iterable[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | None" = None,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | None = None,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> BatchHandle:
    """Enqueue jobs from a lazy iterable in chunks.

    Delegates to :meth:`JobsClient.enqueue_batch_streaming`; see its
    docstring for chunk_size validation and streaming semantics.
    """
    return await self._require_open().enqueue_batch_streaming(
        items,
        batch_id=batch_id,
        connection=connection,
        failure_policy=failure_policy,
        finalizer=finalizer,
        chunk_size=chunk_size,
    )

get_batch `async` ¶

get_batch(batch_id: UUID) -> BatchRow | None

Fetch a single batch row by ID.

Delegates to :meth:JobsClient.get_batch. Returns None when the batch does not exist.

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

async def get_batch(self, batch_id: UUID) -> BatchRow | None:
    """Fetch a single batch row by ID.

    Delegates to :meth:`JobsClient.get_batch`. Returns ``None`` when
    the batch does not exist.
    """
    return await self._require_open().get_batch(batch_id)

list_batches `async` ¶

list_batches(filter: BatchFilter) -> list[BatchSummary]

List batches matching filter, returning :class:BatchSummary objects.

Delegates to :meth:JobsClient.list_batches.

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

async def list_batches(
    self,
    filter: BatchFilter,
) -> list[BatchSummary]:
    """List batches matching *filter*, returning :class:`BatchSummary` objects.

    Delegates to :meth:`JobsClient.list_batches`.
    """
    return await self._require_open().list_batches(filter)

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.

Delegates to :meth:JobsClient.get (jobs table first, then the archive tier; None means the id lives in neither). The result_adapter default and its status-only-idiom rationale are documented there.

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.

    Delegates to :meth:`JobsClient.get` (jobs table first, then the
    archive tier; ``None`` means the id lives in neither). The
    ``result_adapter`` default and its status-only-idiom rationale are
    documented there.
    """
    return await self._require_open().get(job_id, result_adapter=result_adapter)

get_row `async` ¶

get_row(job_id: JobId) -> JobRow | None

Look up a job by id and return the raw JobRow, no handle.

Delegates to :meth:JobsClient.get_row: one backend read, None when the job does not exist, mirroring :meth:get's contract without the handle machinery.

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

async def get_row(self, job_id: JobId) -> JobRow | None:
    """Look up a job by id and return the raw ``JobRow``, no handle.

    Delegates to :meth:`JobsClient.get_row`: one backend read,
    ``None`` when the job does not exist, mirroring :meth:`get`'s
    contract without the handle machinery.
    """
    return await self._require_open().get_row(job_id)

list `async` ¶

list(filter: JobFilter) -> JobPage

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

Delegates to :meth:JobsClient.list, note filter.unfinished selects by terminality ('not yet finished'), not by execution status ('currently executing'). See :class:JobFilter.

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

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

    Delegates to :meth:`JobsClient.list`, note ``filter.unfinished``
    selects by terminality ('not yet finished'), not by execution
    status ('currently executing').  See :class:`JobFilter`.
    """
    return await self._require_open().list(filter)

backpressure `async` ¶

backpressure(
    queues: Sequence[str], *, parent_id: JobId | None = None
) -> BackpressureSnapshot

Read the submit path's backpressure state for queues.

Delegates to :meth:JobsClient.backpressure; see it for the full contract (the depth/cap/children halves, the fail-open unknown state, the advisory semantics). Returns a :class:~taskq.types.BackpressureSnapshot — a read-only typed read mirroring the reader methods above, deliberately not an enqueue kwarg.

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

async def backpressure(
    self,
    queues: Sequence[str],
    *,
    parent_id: "JobId | None" = None,
) -> BackpressureSnapshot:
    """Read the submit path's backpressure state for *queues*.

    Delegates to :meth:`JobsClient.backpressure`; see it for the
    full contract (the depth/cap/children halves, the fail-open
    ``unknown`` state, the advisory semantics). Returns a
    :class:`~taskq.types.BackpressureSnapshot` — a read-only typed
    read mirroring the reader methods above, deliberately not an
    ``enqueue`` kwarg.
    """
    return await self._require_open().backpressure(queues, parent_id=parent_id)

cancel `async` ¶

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

Request cancellation of a job and return a :class:CancelResult.

Delegates to :meth:JobsClient.cancel. The result reports previous_status (the row as first read) and new_status (the row re-read after the cancel request was written; the two are not atomically consistent, a job can move in between), and cancellation_initiated is False when no cancellation was needed (the job was already terminal, or had a cancel request recorded).

Raises :class:KeyError carrying job_id when the job does not exist (never enqueued, or pruned past retention) — the stdlib "asked for an entry by id; it isn't there" idiom, matching :meth:stream and :meth:JobHandle.wait.

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 and return a :class:`CancelResult`.

    Delegates to :meth:`JobsClient.cancel`. The result reports
    ``previous_status`` (the row as first read) and ``new_status``
    (the row re-read after the cancel request was written; the two
    are not atomically consistent, a job can move in between), and
    ``cancellation_initiated`` is ``False`` when no cancellation was
    needed (the job was already terminal, or had a cancel request
    recorded).

    Raises :class:`KeyError` carrying *job_id* when the job does not
    exist (never enqueued, or pruned past retention) — the stdlib
    "asked for an entry by id; it isn't there" idiom, matching
    :meth:`stream` and :meth:`JobHandle.wait`.
    """
    return await self._require_open().cancel(job_id, reason)

cancel_where `async` ¶

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

Cancel all jobs matching filter. See :meth:JobsClient.cancel_where.

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

async def cancel_where(
    self,
    filter: JobFilter,
    reason: str | None = None,
    *,
    allow_empty_filter: bool = False,
) -> BulkCancelResult:
    """Cancel all jobs matching *filter*. See :meth:`JobsClient.cancel_where`."""
    return await self._require_open().cancel_where(
        filter, reason, allow_empty_filter=allow_empty_filter
    )

retry_job `async` ¶

retry_job(job_id: JobId) -> bool

Re-pend a job that has come to rest (a terminal status).

The backend admin operation the admin UI's retry route calls: every terminal status is a valid source (an operator re-run is "run this again", including after a success), non-terminal statuses are refused, and the attempt counter is not reset. Returns False when the job was not in a retryable state, the spent attempt sits at the smallint ceiling, or the job id does not exist — the three cases are indistinguishable here by design (the operation is a single conditional UPDATE); when a False could mean a typo'd id rather than a busy job, check existence first with :meth:get. The same conflict surfaces as a 409 on the admin route.

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

async def retry_job(self, job_id: JobId) -> bool:
    """Re-pend a job that has come to rest (a terminal status).

    The backend admin operation the admin UI's retry route calls:
    every terminal status is a valid source (an operator re-run is
    "run this again", including after a success), non-terminal
    statuses are refused, and the attempt counter is not reset.
    Returns ``False`` when the job was not in a retryable state, the
    spent attempt sits at the smallint ceiling, **or the job id does
    not exist** — the three cases are indistinguishable here by
    design (the operation is a single conditional UPDATE); when a
    ``False`` could mean a typo'd id rather than a busy job, check
    existence first with :meth:`get`. The same conflict surfaces as
    a 409 on the admin route.
    """
    return await self._require_open().backend.retry_job(job_id)

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

    Latency contract
    ----------------
    The first event is the row as it stands when ``stream()`` is
    called. After that:

    * **Redis configured** - the worker publishes every progress and
      state-change event to the job's Redis channel, so a transition
      is delivered as it happens; the row is re-read from Postgres
      every ``poll_timeout`` seconds (default 30) as the safety net.
    * **Postgres only** - nothing announces a job's transitions over
      NOTIFY, so the row is re-read through the client's pool every
      ``min(poll_timeout, 0.5)`` seconds: a terminal write is observed
      within half a second, at the cost of one primary-key read per
      stream per half second. Lower ``poll_timeout`` to tighten it, to
      a floor of 0.1 s; each wait is jittered ±20% so streams opened
      together do not poll in lockstep. No connection is held per
      stream, so this works on a pool-only client as well. A pool or
      connection error on a read is retried on the next poll; reads
      failing for 30 s straight end the stream with
      :class:`~taskq.exceptions.StreamUnavailable`.

    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 — at the opening read, or on any later
        poll whose row has been pruned in the meantime (the stream
        cannot fabricate the promised terminal event, so it raises
        rather than ending quietly).
    StreamUnavailable
        The Postgres poll could not re-read the row for 30 s straight.

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

    Latency contract
    ----------------
    The first event is the row as it stands when ``stream()`` is
    called. After that:

    * **Redis configured** - the worker publishes every progress and
      state-change event to the job's Redis channel, so a transition
      is delivered as it happens; the row is re-read from Postgres
      every ``poll_timeout`` seconds (default 30) as the safety net.
    * **Postgres only** - nothing announces a job's transitions over
      NOTIFY, so the row is re-read through the client's pool every
      ``min(poll_timeout, 0.5)`` seconds: a terminal write is observed
      within half a second, at the cost of one primary-key read per
      stream per half second. Lower ``poll_timeout`` to tighten it, to
      a floor of 0.1 s; each wait is jittered ±20% so streams opened
      together do not poll in lockstep. No connection is held per
      stream, so this works on a pool-only client as well. A pool or
      connection error on a read is retried on the next poll; reads
      failing for 30 s straight end the stream with
      :class:`~taskq.exceptions.StreamUnavailable`.

    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 — at the opening read, or on any later
        poll whose row has been pruned in the meantime (the stream
        cannot fabricate the promised terminal event, so it raises
        rather than ending quietly).
    StreamUnavailable
        The Postgres poll could not re-read the row for 30 s straight.
    """
    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

    gen: AsyncGenerator[JobEvent, None] = (
        _stream_redis(
            self._redis_client,
            self._schema,
            job_id,
            client,
            self._poll_timeout,
            last_seq=row.progress_seq,
            last_status=row.status,
        )
        if self._redis_client is not None
        else _stream_pg(
            job_id,
            client,
            self._poll_timeout,
            last_seq=row.progress_seq,
            last_status=row.status,
        )
    )
    async with contextlib.aclosing(gen) as agen:
        async for evt in agen:
            yield evt
            if evt.terminal:
                return

watch_reclaims `async` ¶

watch_reclaims(
    after_id: int = 0, *, poll_timeout: float | None = None
) -> AsyncIterator[EventRow]

Stream fleet-wide crash-reclaim events as :class:EventRow objects.

Delivery guarantee. At-least-once, and gap-free only under a bounded writer-transaction assumption: an event is silently and permanently missed if a job_events writer transaction stays open longer than reclaim_event_visibility_delay (default 2s , see :data:taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY) between its INSERT and its COMMIT. Ids are allocated at INSERT time but transactions commit out of order, so a late-committing lower-id row can land behind the cursor after the cursor has already advanced past its position; no error is raised anywhere. Sweep and terminal-write transactions are a handful of single-round-trip statements, so 2s is a generous bound under normal operation, but it is an assumption enforced by nothing in the SQL, not a property the query guarantees. So this watcher does not leave detection to chance: on a slow cadence (_VISIBILITY_RISK_CHECK_INTERVAL, 60s) it runs the backend's check_reclaim_visibility_delay_risk diagnostic (when the backend implements it, PostgresBackend does) and logs a loud watch-reclaims-visibility-delay-at-risk warning for every long-open job_events writer it finds. That is a proxy warning, not proof of an actual miss.

Yields job_events rows with kind='state_change' and detail['reason']='lock_expired', ordered by the monotonic event_id cursor ascending. to_state is 'pending' for a retried reclaim, 'crashed' or 'cancelled' (cancel was in-flight when the worker died) for a terminal one. cause names which deadline fired, 'lock_expired' (the worker's global lease) or 'heartbeat_timeout' (the per-job promise) for a leader-sweep reclaim, 'isolate_self' for a reclaim a worker performed on its own jobs when its heartbeat connection died (isolate_self), the same channel so a consumer fanning out on this feed sees heartbeat-loss self-recovery too.

Cursor and duplicate semantics¶

The caller persists the last-seen event_id and passes it back as after_id on resumption. Persist the cursor after processing each event: a crash between processing and persisting re-delivers that event on the next run, delivery is at-least-once, so consumers must dedupe on event_id. The cursor is a watermark, not a reference: pruning job_events rows at or below it is always safe, but rows pruned before the consumer reads them are gone for good, only prune rows older than your slowest consumer's cursor. A cursor far behind after a long outage drains at query speed (full batches are re-polled immediately, not one batch per poll_timeout).

Your own pruning is not the only deleter. The built-in event retention sweep deletes these rows by age alone, cursor position irrelevant: the lock_expired slice is carved out of the ordinary event_retention_period window but only up to :data:~taskq.constants.RECLAIM_OUTBOX_RETENTION_MULTIPLIER times it (100x), after which it is deleted like any other event. The terminal-job prune is a second deleter: its cascade removes the archived jobs' events with them. Both deleters record their highest deleted event id in the event-prune watermark (job_events_prune_state, migration 01.00.20_02) in the same transaction as the delete, and every delivery path of this generator opens on the gap gate: a resumed cursor (a persisted cursor, after_id > 0) strictly below the watermark ends the stream with :class:~taskq.exceptions.EventRetentionGapError before anything is delivered. The signal is conservative (the watermark is a union bound across event kinds: it proves the feed is incomplete, not which slice lost a row), and it closes the silent-loss hole -- a consumer lagging past the horizon would otherwise skip to live with no error on either side, free to believe it saw everything. A fresh watcher (after_id=0) is a new tail and never raises; a cursor at or above the watermark never raises, every deleted id was at or below a position the consumer had already been delivered. Size the retention period against your slowest consumer's worst outage all the same: the gate turns an over-long outage into a loud, actionable end, not into a recovered feed.

Shutdown and backpressure¶

This is a pull-based async generator: events are fetched only as fast as the consumer iterates, so a slow consumer polls slower, no internal buffer grows. To stop, break out of the async for (or cancel the consuming task); generator cleanup removes the LISTEN registration and closes any owned connection.

Transport¶

On Postgres with a LISTEN transport source (dsn, pg_conn_factory, or listen_conn), the method LISTENs on wake_channel(schema) purely as a low-latency wakeup, but always polls backend.poll_reclaim_events(after_id) as the durable source of truth, NOTIFY is an optimisation, never the only path, and a dropped LISTEN connection degrades to (and recovers from) polling automatically. Without a LISTEN transport or on Redis-configured backends, a plain poll loop against poll_reclaim_events runs on poll_timeout.

Usage::

cursor = await load_cursor()  # your own durable store
async for evt in tq.watch_reclaims(after_id=cursor):
    outstanding -= 1          # fan-out completion tracking
    cursor = evt.event_id
    await save_cursor(cursor)  # persist AFTER processing

Raises¶

RuntimeError Called before tq.open() or outside an async with block. EventRetentionGapError The resumed cursor sits behind the event-prune watermark: retention deleted undelivered events, the stream ends instead of silently skipping to live.

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

async def watch_reclaims(
    self,
    after_id: int = 0,
    *,
    poll_timeout: float | None = None,
) -> AsyncIterator[EventRow]:
    """Stream fleet-wide crash-reclaim events as :class:`EventRow` objects.

    **Delivery guarantee.** At-least-once, and gap-free *only under a
    bounded writer-transaction assumption*: an event is **silently and
    permanently missed** if a ``job_events`` writer transaction stays
    open longer than ``reclaim_event_visibility_delay`` (default 2s ,
    see :data:`taskq.constants.RECLAIM_EVENT_VISIBILITY_DELAY`)
    between its INSERT and its COMMIT.  Ids are allocated at INSERT
    time but transactions commit out of order, so a late-committing
    lower-id row can land *behind* the cursor after the cursor has
    already advanced past its position; no error is raised anywhere.
    Sweep and terminal-write transactions are a handful of
    single-round-trip statements, so 2s is a generous bound under
    normal operation, but it is an assumption enforced by nothing in
    the SQL, not a property the query guarantees.  So this watcher
    does not leave detection to chance: on a slow cadence
    (``_VISIBILITY_RISK_CHECK_INTERVAL``, 60s) it runs the backend's
    ``check_reclaim_visibility_delay_risk`` diagnostic (when the
    backend implements it, PostgresBackend does) and logs a loud
    ``watch-reclaims-visibility-delay-at-risk`` warning for every
    long-open ``job_events`` writer it finds.  That is a proxy
    warning, not proof of an actual miss.

    Yields ``job_events`` rows with ``kind='state_change'`` and
    ``detail['reason']='lock_expired'``, ordered by the monotonic
    ``event_id`` cursor ascending.  ``to_state`` is ``'pending'`` for
    a retried reclaim, ``'crashed'`` or ``'cancelled'`` (cancel was
    in-flight when the worker died) for a terminal one.  ``cause``
    names which deadline fired, ``'lock_expired'`` (the worker's
    global lease) or ``'heartbeat_timeout'`` (the per-job promise)
    for a leader-sweep reclaim, ``'isolate_self'`` for a reclaim a
    worker performed on its own jobs when its heartbeat connection
    died (``isolate_self``), the same channel so a consumer fanning
    out on this feed sees heartbeat-loss self-recovery too.

    Cursor and duplicate semantics
    ------------------------------
    The caller persists the last-seen ``event_id`` and passes it back
    as *after_id* on resumption.  Persist the cursor **after**
    processing each event: a crash between processing and persisting
    re-delivers that event on the next run, delivery is
    at-least-once, so consumers must dedupe on ``event_id``.  The
    cursor is a watermark, not a reference: pruning ``job_events``
    rows at or below it is always safe, but rows pruned *before* the
    consumer reads them are gone for good, only prune rows older
    than your slowest consumer's cursor.  A cursor far behind after
    a long outage drains at query speed (full batches are re-polled
    immediately, not one batch per *poll_timeout*).

    Your own pruning is not the only deleter.  The built-in event
    retention sweep deletes these rows by age alone, cursor position
    irrelevant: the ``lock_expired`` slice is carved out of the
    ordinary ``event_retention_period`` window but only up to
    :data:`~taskq.constants.RECLAIM_OUTBOX_RETENTION_MULTIPLIER`
    times it (100x), after which it is deleted like any other event.
    The terminal-job prune is a second deleter: its cascade removes
    the archived jobs' events with them.  Both deleters record their
    highest deleted event id in the event-prune watermark
    (``job_events_prune_state``, migration 01.00.20_02) in the same
    transaction as the delete, and every delivery path of this
    generator opens on the gap gate: a resumed cursor (a persisted
    cursor, ``after_id > 0``) strictly below the watermark ends the
    stream with :class:`~taskq.exceptions.EventRetentionGapError`
    before anything is delivered.  The signal is conservative (the
    watermark is a union bound across event kinds: it proves the feed
    is incomplete, not which slice lost a row), and it closes the
    silent-loss hole -- a consumer lagging past the horizon
    would otherwise skip to live with no error on either side, free to
    believe it saw everything.  A fresh watcher (``after_id=0``) is a
    new tail and never raises; a cursor at or above the watermark
    never raises, every deleted id was at or below a position the
    consumer had already been delivered.  Size the retention period
    against your slowest consumer's worst outage all the same: the
    gate turns an over-long outage into a loud, actionable end, not
    into a recovered feed.

    Shutdown and backpressure
    -------------------------
    This is a pull-based async generator: events are fetched only as
    fast as the consumer iterates, so a slow consumer polls
    slower, no internal buffer grows.  To stop, break out of the
    ``async for`` (or cancel the consuming task); generator cleanup
    removes the LISTEN registration and closes any owned connection.

    Transport
    ---------
    On Postgres with a LISTEN transport source (``dsn``,
    ``pg_conn_factory``, or ``listen_conn``), the method LISTENs on
    ``wake_channel(schema)`` purely as a low-latency wakeup, but
    always polls ``backend.poll_reclaim_events(after_id)`` as the
    durable source of truth, NOTIFY is an optimisation, never the
    only path, and a dropped LISTEN connection degrades to (and
    recovers from) polling automatically.  Without a LISTEN
    transport or on Redis-configured backends, a plain poll loop
    against ``poll_reclaim_events`` runs on ``poll_timeout``.

    Usage::

        cursor = await load_cursor()  # your own durable store
        async for evt in tq.watch_reclaims(after_id=cursor):
            outstanding -= 1          # fan-out completion tracking
            cursor = evt.event_id
            await save_cursor(cursor)  # persist AFTER processing

    Raises
    ------
    RuntimeError
        Called before ``tq.open()`` or outside an ``async with`` block.
    EventRetentionGapError
        The resumed cursor sits behind the event-prune watermark:
        retention deleted undelivered events, the stream ends instead
        of silently skipping to live.
    """
    client = self._require_open()
    timeout = poll_timeout if poll_timeout is not None else self._poll_timeout

    has_listen_source = (
        self._dsn is not None
        or self._pg_conn_factory is not None
        or self._listen_conn is not None
    )
    if self._redis_client is None and has_listen_source:
        gen: AsyncGenerator[EventRow, None] = _watch_reclaims_pg(
            self._dsn,
            self._schema,
            client,
            timeout,
            after_id=after_id,
            pg_conn_factory=self._pg_conn_factory,
            listen_conn=self._listen_conn,
            visibility_delay=(
                self._reclaim_event_visibility_delay
                if self._reclaim_event_visibility_delay is not None
                else RECLAIM_EVENT_VISIBILITY_DELAY
            ),
        )
    else:
        gen = _watch_reclaims_poll(client, timeout, after_id=after_id)

    async with contextlib.aclosing(gen) as agen:
        async for evt in agen:
            yield evt

orjson_response_class ¶

orjson_response_class() -> type[JSONResponse]

Return the orjson-backed :class:starlette.responses.JSONResponse.

Drop-in replacement for starlette's stdlib-json JSONResponse (same constructor, same application/json media type, byte-identical bodies for JSON-representable payloads), bodies are rendered through :func:taskq._json.dumps per the project's never-import-stdlib-json rule. Requires the fastapi extra (starlette) at call time.

Usage with :class:TaskQ, see the :class:JobEvent docstring for the full pattern::

OrjsonJSONResponse = orjson_response_class()

@app.get("/tasks/{job_id}")
async def read_task(job_id: UUID) -> Response:
    return OrjsonJSONResponse(await tq.get_state(job_id))

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

def orjson_response_class() -> "type[JSONResponse]":
    """Return the orjson-backed :class:`starlette.responses.JSONResponse`.

    Drop-in replacement for starlette's stdlib-json ``JSONResponse`` (same
    constructor, same ``application/json`` media type, byte-identical
    bodies for JSON-representable payloads), bodies are rendered through
    :func:`taskq._json.dumps` per the project's never-import-stdlib-json
    rule. Requires the ``fastapi`` extra (starlette) at call time.

    Usage with :class:`TaskQ`, see the :class:`JobEvent` docstring for the
    full pattern::

        OrjsonJSONResponse = orjson_response_class()

        @app.get("/tasks/{job_id}")
        async def read_task(job_id: UUID) -> Response:
            return OrjsonJSONResponse(await tq.get_state(job_id))
    """
    return _orjson_json_response_class()

_jobs ¶

JobsClient, the primary entry point for enqueuing, querying, and cancelling jobs, and managing cron schedules.

Wraps a :class:~taskq.backend._protocol.Backend instance and adds the client-layer behaviours the protocol intentionally omits: payload serialization through the actor's payload_type, CancelResult construction in :meth:cancel, typed :class:JobHandle[R] wrapping in :meth:enqueue / :meth:get, and cron schedule management via :meth:create_schedule, :meth:list_schedules, :meth:update_schedule, :meth:delete_schedule.

The backend is injected at construction so the same client can target either an :class:~taskq.testing.in_memory.InMemoryBackend (tests) or a :class:taskq.backend.postgres.PostgresBackend (production).

all `module-attribute` ¶

__all__ = ['JobsClient']

logger `module-attribute` ¶

logger: BoundLogger = structlog.get_logger(__name__)

JobsClient ¶

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

Public API for job operations.

Delegates to the injected :class:Backend and wraps results in typed :class:JobHandle[R] instances. The client owns the payload-serialization step that turns a typed P into the dict[str, object] carried by :class:EnqueueArgs; the backend sees only erased payloads.

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

def __init__(
    self,
    backend: Backend,
    *,
    clock: Clock | None = None,
    settings: "TaskQSettings | None" = None,
    capacity_cache_ttl: float = DEFAULT_CAPACITY_CACHE_TTL,
) -> 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._unique_for_warner = UniqueForNoIdentityWarner()
    self._capacity_cache = ActorCapacityCache(
        backend,
        ttl=capacity_cache_ttl,
        # A bare construction (no settings) invents no knob: strict
        # off, the fail-open posture the note's own tests pin.
        # getattr-with-default: the client-layer tests drive bare
        # stand-in namespaces for settings (the notify_enabled
        # convention), strict defaults False there. env_queues is the
        # two-source rule's second source — this process's own
        # declared set, when it has one (a base-settings client has
        # none; the snapshot is then the only source, the
        # documented web-without-env corner).
        queues_strict=bool(getattr(settings, "queues_strict", False)),
        env_queues=getattr(settings, "queues", None),
    )
    # Why resolved here: every enqueue path in this client validates
    # against one number, and a client built without settings still gets
    # the shipped default rather than a second literal.
    self._idempotency_max_bytes: int = (
        settings.idempotency_key_max_bytes
        if settings is not None
        else MAX_IDEMPOTENCY_KEY_BYTES
    )

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

invalidate_actor_capacity_cache ¶

invalidate_actor_capacity_cache() -> None

Drop the cached actor_config.max_pending snapshot.

The next enqueue refreshes from the backend instead of waiting out the TTL. Not needed in normal operation (staleness is bounded by capacity_cache_ttl, default 5s); intended for tests and for tooling that knows it just changed the table and cannot wait out the TTL.

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

def invalidate_actor_capacity_cache(self) -> None:
    """Drop the cached ``actor_config.max_pending`` snapshot.

    The next enqueue refreshes from the backend instead of waiting
    out the TTL. Not needed in normal operation (staleness is
    bounded by ``capacity_cache_ttl``, default 5s); intended for
    tests and for tooling that knows it just changed the table and
    cannot wait out the TTL.
    """
    self._capacity_cache.invalidate()

enqueue `async` ¶

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

Enqueue a job for the given actor and return a typed handle.

The payload is serialized through ref.payload_type so the EnqueueArgs.payload carried over the backend boundary is a plain dict[str, object] ready for the JSONB column. The returned :class:JobHandle[R] carries ref.result_adapter so :meth:JobHandle.wait can validate the stored result back to R.

The actor's name is NOT validated against the registration table here: any :class:~taskq.actor.ActorRef is accepted, and a ref whose actor no worker fleet declares enqueues successfully and then parks (every worker that claims it snoozes it back with released_reason: "actor-not-found", budget-free; the stranded-jobs detector surfaces it). If "my job never runs", check the actor spelling against the worker's registry and taskq job show's released reason before looking at queues or capacity. This is the same defer-to-fire-time doctrine :meth:create_schedule documents.

The metadata.singleton key is reserved by the library for singleton enforcement. When ref.singleton is True the library unconditionally writes metadata.singleton = True, overriding any caller-supplied value. Callers MUST NOT set metadata.singleton manually.

max_pending:

When the actor's effective max_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.
The effective limit is operator-owned: a non-NULL stored actor_config.max_pending (set via taskq actor-config set --max-pending) wins over the @actor(max_pending=...) literal; a cleared or absent stored value falls back to the literal. The client reads the stored value through a TTL-bounded cache (default 5s staleness; see :class:taskq.client._capacity.ActorCapacityCache), so an operator change takes effect fleet-wide within seconds without any redeploy or restart.
Evaluation order at enqueue: unique_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 unique within its idempotency_scope (composite (idempotency_scope, idempotency_key) uniqueness). The default scope (idempotency_scope=None or "") preserves the prior global-until-prune behavior exactly, so existing callers see zero behavior change. Passing an explicit scope (e.g. a run/batch/epoch id) lets two enqueues with the same business key in different scopes both succeed, decoupling the dedupe horizon from prune_retention_*.
Key length is bounded at idempotency_key_max_bytes (TASKQ_IDEMPOTENCY_KEY_MAX_BYTES, default 1024 UTF-8 bytes) , the bound is the composite unique index's btree entry size, not a round number. Empty and whitespace-only keys raise :class:ValueError at the client boundary before any backend call. The same bound applies to idempotency_scope; an empty scope ("") is valid and equivalent to None (the default/global scope).
No time-based (TTL) dedupe window. idempotency_scope decouples the dedupe horizon from prune_retention_* by namespace, not by time, there is no idempotency_ttl or equivalent "dedupe for the next N seconds" parameter. A key within a given scope still dedupes until pruned, exactly like the pre-scope global behavior, just scoped to that namespace. This is a deliberate scope decision, not an oversight: a real sliding-window TTL cannot be expressed as a single static unique index the way scope can, a sliding window would require either abandoning the atomic INSERT ... ON CONFLICT for a check-then-insert lock (weaker concurrency guarantee) or encoding time-bucketing into the key itself (coarser, edge-artifact-prone semantics). If your use case genuinely needs "dedupe for the next hour, not forever," encode the window into the scope yourself (e.g. a time-bucketed scope string) until/unless a TTL parameter ships as a separate feature.
Rolling-deploy note: if this schema is mid-upgrade (the 01.00.03_01_pre_idempotency_scope.sql migration applied but 01.00.03_01_post_idempotency_scope_drop_old_index.sql not yet applied), reusing the same idempotency_key under two different idempotency_scope values raises :class:~taskq.exceptions.ScopedIdempotencyMigrationPendingError rather than silently dedupe against the wrong scope's job. The trigger is a key existing under a different scope, in either direction, an unscoped call reusing a key first written under a non-default scope raises it too. Only brand-new keys and same-scope repeats are unaffected. See that exception's docstring and the migration file's header comment for the full rationale.

unique_for:

unique_for deduplication is serialized, not best-effort: the preflight and the INSERT run under one transaction-scoped advisory lock on (actor, identity_key), so concurrent enqueues of the same identity produce exactly one job and every caller is handed that job. This holds on pool connections and on a caller-supplied connection with no open transaction. On a caller-owned OPEN transaction the lock spans that transaction instead, so single-flight holds until its commit/rollback, a long-lived caller transaction can exhaust another same-identity enqueue's bounded wait (:class:~taskq.exceptions.UniqueForLockTimeoutError) rather than dedup against it. The dispatch CTE's running_identities filter remains the execution-level guard behind the enqueue-level one.
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.

allow_unregistered:

The per-submit escape from TASKQ_QUEUES_STRICT: with strict on, a queue no registered actor's assignment routes to raises :class:~taskq.exceptions.UnknownQueueError before any write; allow_unregistered=True skips that verdict for THIS call — for genuinely dynamic queue names (tenant-prefixed, sharded at runtime) that no registration can predict. The unserved-queue NOTE still fires (the diagnostic sits below the gate), so a typo'd name stays visible even on an escaped call. With strict off the flag changes nothing. The batch arms carry no per-item flag: their queues are the items' actors' own declared routes, the thing registration populates; use single enqueues for dynamic names.

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,
    allow_unregistered: bool = False,
    scheduled_at: datetime | None = None,
    priority: int | None = None,
    schedule_to_close: datetime | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
    identity_key: IdentityKey | None = None,
    fairness_key: str | None = None,
    idempotency_key: IdempotencyKey | None = None,
    idempotency_scope: str | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    metadata: dict[str, object] | None = None,
    tags: list[str] | None = None,
) -> JobHandle[R]:
    """Enqueue a job for the given actor and return a typed handle.

    The payload is serialized through ``ref.payload_type`` so the
    ``EnqueueArgs.payload`` carried over the backend boundary is a
    plain ``dict[str, object]`` ready for the JSONB column. The
    returned :class:`JobHandle[R]` carries ``ref.result_adapter`` so
    :meth:`JobHandle.wait` can validate the stored result back to
    ``R``.

    The actor's name is NOT validated against the registration table
    here: any :class:`~taskq.actor.ActorRef` is accepted, and a ref
    whose actor no worker fleet declares enqueues successfully and
    then parks (every worker that claims it snoozes it back with
    ``released_reason: "actor-not-found"``, budget-free; the
    stranded-jobs detector surfaces it). If "my job never runs",
    check the actor spelling against the worker's registry and
    ``taskq job show``'s released reason before looking at queues or
    capacity. This is the same defer-to-fire-time doctrine
    :meth:`create_schedule` documents.

    The ``metadata.singleton`` key is reserved by the library for
    singleton enforcement. When ``ref.singleton`` is ``True`` the
    library unconditionally writes ``metadata.singleton = True``,
    overriding any caller-supplied value. Callers MUST NOT set
    ``metadata.singleton`` manually.

    **max_pending:**

    - When the actor's effective ``max_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.

    - The effective limit is **operator-owned**: a non-NULL stored
      ``actor_config.max_pending`` (set via
      ``taskq actor-config set --max-pending``) wins over the
      ``@actor(max_pending=...)`` literal; a cleared or absent
      stored value falls back to the literal. The client reads the
      stored value through a TTL-bounded cache (default 5s
      staleness; see :class:`taskq.client._capacity.ActorCapacityCache`),
      so an operator change takes effect fleet-wide within seconds
      without any redeploy or restart.

    - Evaluation order at enqueue: ``unique_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 unique within its ``idempotency_scope``
      (composite ``(idempotency_scope, idempotency_key)`` uniqueness).
      The default scope (``idempotency_scope=None`` or ``""``) preserves
      the prior global-until-prune behavior exactly, so existing callers
      see zero behavior change. Passing an explicit scope (e.g. a
      run/batch/epoch id) lets two enqueues with the same business key
      in different scopes both succeed, decoupling the dedupe horizon
      from ``prune_retention_*``.

    - Key length is bounded at ``idempotency_key_max_bytes``
      (``TASKQ_IDEMPOTENCY_KEY_MAX_BYTES``, default 1024 UTF-8 bytes) ,
      the bound is the composite unique index's btree entry size, not a
      round number. Empty and whitespace-only keys raise
      :class:`ValueError` at the client boundary before any backend
      call. The same bound applies to ``idempotency_scope``; an empty
      scope (``""``) is valid and equivalent to ``None`` (the
      default/global scope).

    - **No time-based (TTL) dedupe window.** ``idempotency_scope``
      decouples the dedupe horizon from ``prune_retention_*`` by
      namespace, not by time, there is no ``idempotency_ttl`` or
      equivalent "dedupe for the next N seconds" parameter. A key
      within a given scope still dedupes **until pruned**, exactly
      like the pre-scope global behavior, just scoped to that
      namespace. This is a deliberate scope decision, not an
      oversight: a real sliding-window TTL cannot be expressed as a
      single static unique index the way scope can, a sliding
      window would require either abandoning the atomic ``INSERT
      ... ON CONFLICT`` for a check-then-insert lock (weaker
      concurrency guarantee) or encoding time-bucketing into the key
      itself (coarser, edge-artifact-prone semantics). If your use
      case genuinely needs "dedupe for the next hour, not forever,"
      encode the window into the scope yourself (e.g. a
      time-bucketed scope string) until/unless a TTL parameter ships
      as a separate feature.

    - Rolling-deploy note: if this schema is mid-upgrade (the
      ``01.00.03_01_pre_idempotency_scope.sql`` migration applied but
      ``01.00.03_01_post_idempotency_scope_drop_old_index.sql`` not
      yet applied), reusing the same ``idempotency_key`` under two
      *different* ``idempotency_scope`` values raises
      :class:`~taskq.exceptions.ScopedIdempotencyMigrationPendingError`
      rather than silently dedupe against the wrong scope's job. The
      trigger is a key existing under a different scope, in *either*
      direction, an unscoped call reusing a key first written under
      a non-default scope raises it too. Only brand-new keys and
      same-scope repeats are unaffected. See that exception's
      docstring and the migration file's header comment for the full
      rationale.

    **unique_for:**

    - ``unique_for`` deduplication is **serialized**, not best-effort:
      the preflight and the INSERT run under one transaction-scoped
      advisory lock on ``(actor, identity_key)``, so concurrent
      enqueues of the same identity produce exactly one job and every
      caller is handed that job. This holds on pool connections and on
      a caller-supplied connection with no open transaction. On a
      caller-owned OPEN transaction the lock spans that transaction
      instead, so single-flight holds until its commit/rollback, a
      long-lived caller transaction can exhaust another same-identity
      enqueue's bounded wait
      (:class:`~taskq.exceptions.UniqueForLockTimeoutError`) rather
      than dedup against it. The dispatch CTE's ``running_identities``
      filter remains the execution-level guard behind the
      enqueue-level one.

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

    **allow_unregistered:**

    - The per-submit escape from ``TASKQ_QUEUES_STRICT``: with strict
      on, a queue no registered actor's assignment routes to raises
      :class:`~taskq.exceptions.UnknownQueueError` before any write;
      ``allow_unregistered=True`` skips that verdict for THIS call —
      for genuinely dynamic queue names (tenant-prefixed, sharded at
      runtime) that no registration can predict. The unserved-queue
      NOTE still fires (the diagnostic sits below the gate), so a
      typo'd name stays visible even on an escaped call. With strict
      off the flag changes nothing. The batch arms carry no per-item
      flag: their queues are the items' actors' own declared routes,
      the thing registration populates; use single enqueues for
      dynamic names.
    """
    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,
    ):
        effective_max_pending = await self._capacity_cache.effective_max_pending(
            ref.name, ref.max_pending
        )
        # Zero-I/O verdict from the snapshot the resolve just refreshed:
        # a queue no stored assignment routes to strands the row
        # silently (pending forever), the note is the enqueue-time
        # signal, warn-once per queue per TTL. Under
        # TASKQ_QUEUES_STRICT this same verdict REFUSES
        # (UnknownQueueError) instead of warning — unless
        # allow_unregistered=True, the per-submit escape for a
        # genuinely dynamic queue name (the note still fires on an
        # escaped call, so a typo stays visible).
        self._capacity_cache.maybe_warn_unserved_queue(
            resolved_queue, actor=ref.name, allow_unregistered=allow_unregistered
        )
        # An explicit trace_id/span_id overrides the ambient span, per
        # docs/guides/jobs-clients.md: "pass explicitly to override or
        # to propagate an external trace context". Explicit values
        # dropped in favour of the extracted ones leave cross-service
        # propagation silently producing an unlinked
        # consumer span. SubJobEnqueuer.enqueue has no such override and
        # correctly exposes no parameter for one.
        args = build_enqueue_args(
            ref,
            payload,
            queue=queue,
            scheduled_at=scheduled_at,
            priority=priority,
            fairness_key=fairness_key,
            metadata=metadata,
            parent_id=current_parent_id(),
            identity_key=identity_key,
            idempotency_key=idempotency_key,
            idempotency_scope=idempotency_scope,
            trace_id=trace_id if trace_id is not None else extracted_trace_id,
            span_id=span_id if span_id is not None else extracted_span_id,
            schedule_to_close=schedule_to_close,
            start_to_close=start_to_close,
            heartbeat_timeout=heartbeat_timeout,
            max_pending=effective_max_pending,
            tags=tags,
            idempotency_max_bytes=self._idempotency_max_bytes,
        )
        if span.is_recording():
            # Why the guard: on a non-recording span (no SDK, sampling)
            # set_attribute discards the value, so the str() of the job
            # id is paid per enqueue for nothing. Skipped, the exported
            # spans are unchanged: a recording span still gets exactly
            # this attribute.
            span.set_attribute("messaging.message.id", str(args.id))
        if args.unique_for is not None and args.identity_key is None:
            self._unique_for_warner.maybe_warn(
                actor=ref.name, queue=ref.queue, unique_for=args.unique_for
            )
        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=str(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,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | None = None,
) -> BatchHandle

Enqueue multiple jobs in a single batched INSERT and return a :class:~taskq.batch.BatchHandle.

All items share a single batch_id UUID written into each job's metadata.batch_id field (as a string). When batch_id is not supplied it is auto-generated as a UUIDv7 via :func:~taskq._ids.new_job_id.

failure_policy:

When failure_policy is set (e.g. :class:~taskq.batch_policy.AbortBatchAfter), a batches row is created with the policy's failure threshold. After each child job reaches a terminal state, the :func:~taskq.batch.apply_batch_terminal_outcome hook inspects the outcome: succeeded resets the consecutive-failure counter; failed increments it and aborts the batch if the threshold is reached. Aborting cancels all pending and scheduled child jobs and sets the batch row to aborted.

finalizer:

When finalizer is set, a finalizer job is enqueued alongside the batch. The finalizer is NOT stamped with batch_id metadata (deadlock prevention, if it were, wait_for_batch would count it as a child and the finalizer would wait for itself). The batch row's finalizer_job_id column records the link, and wait_for_batch automatically excludes that job from counts. The finalizer is dispatched immediately; the in-actor wait_for_batch snooze pattern gates on child-job completion.

Transactional enqueue:

When failure_policy or finalizer is set and connection is None, the entire operation (batch row + all child jobs + finalizer) is inserted in a single transaction via :meth:Backend.enqueue_batch_atomic. If any insert fails, no rows are committed. When a connection is provided, the caller controls the transaction boundary; the batch row and finalizer are created as the last statements on that connection.

Validation rules:

len(items) == 0 raises :class:ValueError.
len(items) > MAX_BATCH_SIZE 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:

Admission is partitioned per actor. The backend issues one aggregated SELECT actor, count(*) … WHERE actor = ANY($1) GROUP BY actor for the entire batch, resolves per-actor effective limits (operator-owned stored value when set, else the @actor(...) literal, same resolution as :meth:enqueue), and admits every within-cap actor's items; an over-cap actor's items are refused as a whole group (never partially filled up to the cap). When any actor is refused, the within-cap actors' items are still enqueued and :class:~taskq.exceptions.BatchMaxPendingExceededError raises afterwards, it names each refused actor, the refused item indices into items, and the admitted count, so a retry can target only the refused items (or rely on idempotency_keys to deduplicate a whole-batch retry). The exception is deliberately not a :class:~taskq.exceptions.MaxPendingExceededError subclass: handlers for that type assume nothing was enqueued, and under this error part of the batch is already stored.

When failure_policy or finalizer is set and connection is None (the atomic path), a cap violation keeps the all-or-nothing contract instead: the whole single transaction rolls back and plain :class:~taskq.exceptions.MaxPendingExceededError raises with nothing committed.

On a caller-supplied connection with an open transaction, the admitted items are inserted but their durability follows that transaction's commit/rollback.

unique_for (not applied on batch paths):

Actor-declared unique_for is a single-enqueue contract (see :meth:enqueue). The batch INSERT writes every item without a per-identity preflight, the single path's advisory-lock + preflight round trips are exactly what bulk throughput exists to avoid, so two batch items with the same (actor, identity_key) both insert, and the dispatch CTE's identity serialization ensures only one runs at a time. unique_for items are conservatively fully counted toward max_pending (a batch mixing unique_for retries near the cap may refuse loudly rather than admit silently). Deduplicate batch items with per-item idempotency_keys, that arbiter IS applied, below. Whether the batch tier should honor unique_for is an open design decision; this documents current behavior.

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,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | None = None,
) -> BatchHandle:
    """Enqueue multiple jobs in a single batched INSERT and return a
    :class:`~taskq.batch.BatchHandle`.

    All ``items`` share a single ``batch_id`` UUID written into each
    job's ``metadata.batch_id`` field (as a string).  When
    ``batch_id`` is not supplied it is auto-generated as a UUIDv7 via
    :func:`~taskq._ids.new_job_id`.

    **failure_policy:**

    When ``failure_policy`` is set (e.g.
    :class:`~taskq.batch_policy.AbortBatchAfter`), a ``batches`` row
    is created with the policy's failure threshold. After each child
    job reaches a terminal state, the
    :func:`~taskq.batch.apply_batch_terminal_outcome` hook inspects
    the outcome: ``succeeded`` resets the consecutive-failure
    counter; ``failed`` increments it and aborts the batch if the
    threshold is reached. Aborting cancels all pending and scheduled
    child jobs and sets the batch row to ``aborted``.

    **finalizer:**

    When ``finalizer`` is set, a finalizer job is enqueued alongside
    the batch. The finalizer is NOT stamped with ``batch_id``
    metadata (deadlock prevention, if it were, ``wait_for_batch``
    would count it as a child and the finalizer would wait for
    itself). The batch row's ``finalizer_job_id`` column records the
    link, and ``wait_for_batch`` automatically excludes that job
    from counts. The finalizer is dispatched immediately; the
    in-actor ``wait_for_batch`` snooze pattern gates on child-job
    completion.

    **Transactional enqueue:**

    When ``failure_policy`` or ``finalizer`` is set and
    ``connection`` is ``None``, the entire operation (batch row +
    all child jobs + finalizer) is inserted in a single transaction
    via :meth:`Backend.enqueue_batch_atomic`. If any insert fails,
    no rows are committed. When a ``connection`` is provided, the
    caller controls the transaction boundary; the batch row and
    finalizer are created as the last statements on that connection.

    **Validation rules:**

    - ``len(items) == 0`` raises :class:`ValueError`.
    - ``len(items) > MAX_BATCH_SIZE`` 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:**

    Admission is partitioned per actor.  The backend issues one
    aggregated ``SELECT actor, count(*) … WHERE actor = ANY($1)
    GROUP BY actor`` for the entire batch, resolves per-actor
    effective limits (operator-owned stored value when set, else the
    ``@actor(...)`` literal, same resolution as :meth:`enqueue`),
    and admits every within-cap actor's items; an over-cap actor's
    items are refused as a whole group (never partially filled up to
    the cap).  When any actor is refused, the within-cap actors'
    items are still enqueued and
    :class:`~taskq.exceptions.BatchMaxPendingExceededError` raises
    afterwards, it names each refused actor, the refused item
    indices into ``items``, and the admitted count, so a retry can
    target only the refused items (or rely on ``idempotency_key``s
    to deduplicate a whole-batch retry).  The exception is
    deliberately not a :class:`~taskq.exceptions.MaxPendingExceededError`
    subclass: handlers for that type assume nothing was enqueued,
    and under this error part of the batch is already stored.

    When ``failure_policy`` or ``finalizer`` is set and
    ``connection`` is ``None`` (the atomic path), a cap violation
    keeps the all-or-nothing contract instead: the whole single
    transaction rolls back and plain
    :class:`~taskq.exceptions.MaxPendingExceededError` raises with
    nothing committed.

    On a caller-supplied connection with an open transaction, the
    admitted items are inserted but their durability follows that
    transaction's commit/rollback.

    **unique_for (not applied on batch paths):**

    Actor-declared ``unique_for`` is a single-enqueue contract (see
    :meth:`enqueue`). The batch INSERT writes every item without a
    per-identity preflight, the single path's advisory-lock +
    preflight round trips are exactly what bulk throughput exists to
    avoid, so two batch items with the same
    ``(actor, identity_key)`` both insert, and the dispatch CTE's
    identity serialization ensures only one runs at a time.
    ``unique_for`` items are conservatively fully counted toward
    ``max_pending`` (a batch mixing ``unique_for`` retries near the
    cap may refuse loudly rather than admit silently). Deduplicate
    batch items with per-item ``idempotency_key``s, that arbiter IS
    applied, below. Whether the batch tier should honor
    ``unique_for`` is an open design decision; this documents
    current behavior.

    **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) > MAX_BATCH_SIZE:
        raise ValueError(
            f"items must contain at most {MAX_BATCH_SIZE} 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
        validate_actor_payload(ref.payload_type, item.payload, actor=ref.name)
        validate_idempotency(
            item.idempotency_key,
            item.idempotency_scope,
            self._idempotency_max_bytes,
            where=f" for item {i}",
        )

    # Resolve the effective cap per actor (stored operator value when
    # set, else the ``@actor(...)`` literal, same resolution as
    # :meth:`enqueue`) so the per-item args carry it into the backend,
    # whose admission check is the single enforcement point: it counts
    # live (not through this client's TTL cache), discounts idempotency
    # pairs that will dedupe instead of writing, and partitions
    # admission per actor. The old client-side aggregated pre-check
    # raised here for the WHOLE call, one capped actor aborted
    # everyone's items, and its count was strictly less
    # informed than the backend's, so it was removed rather than
    # duplicated.
    effective_mp: dict[str, int | None] = {}
    # Why memoized by ACTOR NAME and verdict-check per new name only:
    # this assumes name -> queue is a FUNCTION (two refs sharing a
    # name always share a queue). That invariant is enforced at the
    # registry boundary (worker/_bootstrap refuses registry keys that
    # disagree with ref.name, and sync_actor_config would
    # CardinalityViolation on a duplicate name), so per-name
    # memoization cannot hide a second queue behind one actor.
    for item in items:
        ref = item.actor_ref
        if ref.name not in effective_mp:
            effective_mp[ref.name] = await self._capacity_cache.effective_max_pending(
                ref.name, ref.max_pending
            )
            # Same snapshot verdict as the single arm, warn-once per
            # queue per TTL; batch items carry no per-item queue, the
            # ref's own assignment is what the row is written to.
            self._capacity_cache.maybe_warn_unserved_queue(ref.queue, actor=ref.name)

    # Build per-item EnqueueArgs carrying the resolved limits for the
    # backend's per-actor admission check. The ambient parent context
    # is read ONCE per call (LIB-2): every item is a child of the
    # same parent when one is set.
    args_list = build_batch_args(
        items,
        resolved_batch_id,
        max_pending_by_actor=effective_mp,
        parent_id=current_parent_id(),
    )

    queue = items[0].actor_ref.queue
    has_batch_extras = failure_policy is not None or finalizer is not None

    # Build finalizer EnqueueArgs (without batch_id stamping, deadlock prevention).
    finalizer_args: EnqueueArgs | None = None
    if finalizer is not None:
        # The finalizer rides the SAME strict verdict as the items: the
        # per-item loop above iterates items only, and the finalizer's
        # args write through BOTH sub-paths below (the atomic arm and
        # the caller-connection arm), so without this the finalizer
        # was the strict bypass — the exact orphan class strict exists
        # to kill (its row stored on an unregistered queue, no
        # refusal, no note). The snapshot is warm here — the awaited
        # cap resolution in the loop spent the one refresh — so the
        # verdict is a pure snapshot lookup, zero extra I/O. One
        # verdict covers both write sites: the args are built once,
        # after it.
        self._capacity_cache.maybe_warn_unserved_queue(
            finalizer.actor_ref.queue, actor=finalizer.actor_ref.name
        )
        finalizer_args = build_enqueue_args(
            finalizer.actor_ref,
            finalizer.payload,
            scheduled_at=finalizer.scheduled_at,
            priority=finalizer.priority,
            fairness_key=finalizer.fairness_key,
            identity_key=finalizer.identity_key,
            idempotency_key=finalizer.idempotency_key,
            idempotency_scope=finalizer.idempotency_scope,
            metadata=dict(finalizer.metadata),
            start_to_close=finalizer.start_to_close,
            tags=finalizer.tags,
            idempotency_max_bytes=self._idempotency_max_bytes,
            parent_id=current_parent_id(),
        )

    # Build BatchRow when failure_policy OR finalizer is set (C3:
    # finalizer-only batches also need a row for list_batches
    # discoverability and finalizer_job_id auto-exclusion in
    # wait_for_batch). When only finalizer is set, failure_threshold=None.
    batch_row: BatchRow | None = None
    if failure_policy is not None or finalizer is not None:
        threshold = failure_policy.failure_threshold if failure_policy is not None else None
        batch_row = BatchRow(
            id=resolved_batch_id,
            queue=queue,
            status="active",
            expected_size=len(items),
            consecutive_failures=0,
            failure_threshold=threshold,
            finalizer_job_id=finalizer_args.id if finalizer_args is not None else None,
            originating_actor=None,
            created_at=self._clock.now(),
            completed_at=None,
            metadata={},
        )

    if has_batch_extras and connection is None:
        # Autonomous atomic path: delegate to backend.enqueue_batch_atomic.
        all_rows = await self._backend.enqueue_batch_atomic(
            args_list,
            batch_id=resolved_batch_id,
            queue=queue,
            batch_row=batch_row,
            finalizer_args=finalizer_args,
        )
        rows = all_rows[: len(items)]
        finalizer_row = all_rows[len(items) :][0] if finalizer is not None else None
    else:
        # Caller-owned transaction or no extras: use regular enqueue_batch.
        # M3: create the batch row BEFORE job inserts so the row exists
        # when the first terminal write triggers the batch hook (on an
        # autocommit conn, jobs can dispatch and fail before the row
        # exists otherwise). Insert the finalizer first so its returned
        # row id is known for finalizer_job_id (M4: idempotency collision
        # may return a different id than finalizer_args.id).
        finalizer_row = None
        if finalizer is not None:
            assert finalizer_args is not None
            finalizer_row = await self._backend.enqueue_with_conn(connection, finalizer_args)  # type: ignore[arg-type]  # Why: guarded by has_batch_extras; when connection is provided it is runtime-compatible
        if batch_row is not None:
            if finalizer_row is not None:
                batch_row = replace(batch_row, finalizer_job_id=finalizer_row.id)
            await self._backend.create_batch(
                batch_row.id,
                batch_row.queue,
                batch_row.expected_size,
                batch_row.failure_threshold,
                batch_row.finalizer_job_id,
                batch_row.originating_actor,
                connection=connection,  # type: ignore[arg-type]  # Why: connection may be None but create_batch handles that
            )
        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)

    # Build finalizer handle (if any) and append to job_handles for backward compat.
    finalizer_handle: JobHandle[BaseModel | None] | None = None
    if finalizer is not None and finalizer_row is not None:
        finalizer_handle = JobHandle(
            client=self,
            row=finalizer_row,
            result_adapter=finalizer.actor_ref.result_adapter,
            was_existing=(finalizer_row.id != finalizer_args.id)
            if finalizer_args is not None
            else False,
            _redis_client=self._redis_client,
            _settings=self._settings,
        )
        handles.append(finalizer_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),
        finalizer_handle=finalizer_handle,
    )

enqueue_batch_streaming `async` ¶

enqueue_batch_streaming(
    items: Iterable[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: Connection | None = None,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | None = None,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> BatchHandle

Enqueue jobs from a lazy iterable in chunks, returning a single :class:~taskq.batch.BatchHandle.

Unlike :meth:enqueue_batch, this method accepts an :class:~collections.abc.Iterable (including generators) and inserts in chunks of chunk_size (1-MAX_BATCH_SIZE). All items share the same batch_id. Payloads are validated on the fly as each chunk is built.

When failure_policy or finalizer is set and connection is None, the entire operation is delegated to :meth:Backend.enqueue_batch_atomic for single-transaction atomicity. Otherwise, chunks are inserted via :meth:Backend.enqueue_batch: with a caller-supplied connection all chunks share that connection's open transaction (one transaction aggregate, the caller owns the boundary); with NO connection each chunk is its own pool transaction, so progress is committed incrementally, chunk by chunk.

No-connection failure surface: when chunk N fails (cap refusal, payload validation, a driver error), chunks 1..N-1 are already durably committed and nothing is returned, the BatchHandle is only constructed after the whole stream drains. A blind retry of the full iterable would duplicate the committed prefix. Retry safely by giving items idempotency_keys (a retry deduplicates against the committed rows) or by resuming from the point of failure.

Failure-policy counting limitation (caller-connection path): the batch row is created AFTER all chunk inserts, it must carry the final expected_size and create_batch is INSERT, not upsert, so a child job enqueued on that connection that reaches a terminal state BEFORE the row exists is not counted toward failure_policy: increment_batch_failures finds no row and returns (0, None, 0). The atomic (no-connection) path is unaffected, its single transaction makes the batch row and the child jobs visible together.

max_pending: enforced per chunk with the same per-actor partition as :meth:enqueue_batch, each chunk's items carry their actors' effective caps (operator-stored override when set, else the @actor(...) literal, same resolution as :meth:enqueue), the backend admits every within-cap actor's items, and an over-cap actor's items are refused. On the caller-connection path sequential chunks share one transaction, so per-chunk admission accumulates to the true aggregate. On the no-connection path each chunk counts against the rows the previous committed chunks wrote, which also accumulates to the true aggregate, but the refusal surfaces after the committed prefix: :class:~taskq.exceptions.BatchMaxPendingExceededError raises with STREAM-GLOBAL refused item indices and an admitted_count covering every committed item; items after the refusing chunk were never attempted. Retry only the refused items (or rely on idempotency_keys). The atomic path (failure_policy/finalizer, no connection) resolves the same effective caps and keeps all-or-nothing: a cap violation rolls back the entire single transaction and raises plain :class:~taskq.exceptions.MaxPendingExceededError with nothing committed.

unique_for (not applied on batch paths): actor-declared unique_for is a single-enqueue contract (see :meth:enqueue_batch's disclosure), every item writes, items sharing an (actor, identity_key) are not deduplicated, and unique_for items are fully counted toward max_pending. This holds on both arms (the per-chunk INSERT and the atomic path's chunked INSERT). The idempotency_key arbiter IS applied on both arms: a colliding item returns the existing row, exactly as :meth:enqueue_batch does.

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

async def enqueue_batch_streaming(
    self,
    items: Iterable[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | None" = None,
    failure_policy: BatchFailurePolicy | None = None,
    finalizer: EnqueueItem | None = None,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> BatchHandle:
    """Enqueue jobs from a lazy iterable in chunks, returning a single
    :class:`~taskq.batch.BatchHandle`.

    Unlike :meth:`enqueue_batch`, this method accepts an
    :class:`~collections.abc.Iterable` (including generators) and
    inserts in chunks of ``chunk_size`` (1-MAX_BATCH_SIZE).  All items share
    the same ``batch_id``.  Payloads are validated on the fly as
    each chunk is built.

    When ``failure_policy`` or ``finalizer`` is set and
    ``connection`` is ``None``, the entire operation is delegated to
    :meth:`Backend.enqueue_batch_atomic` for single-transaction
    atomicity.  Otherwise, chunks are inserted via
    :meth:`Backend.enqueue_batch`: with a caller-supplied connection
    all chunks share that connection's open transaction (one
    transaction aggregate, the caller owns the boundary); with NO
    connection each chunk is its own pool transaction, so progress
    is committed incrementally, chunk by chunk.

    **No-connection failure surface:** when chunk *N* fails (cap
    refusal, payload validation, a driver error), chunks 1..*N*-1
    are already durably committed and nothing is returned, the
    ``BatchHandle`` is only constructed after the whole stream
    drains.  A blind retry of the full iterable would duplicate the
    committed prefix.  Retry safely by giving items
    ``idempotency_key``s (a retry deduplicates against the committed
    rows) or by resuming from the point of failure.

    **Failure-policy counting limitation (caller-connection path):**
    the batch row is created AFTER all chunk inserts, it must carry
    the final ``expected_size`` and ``create_batch`` is INSERT, not
    upsert, so a child job enqueued on that connection that
    reaches a terminal state BEFORE the row exists is not counted
    toward ``failure_policy``: ``increment_batch_failures`` finds
    no row and returns ``(0, None, 0)``. The atomic
    (no-connection) path is unaffected, its single transaction
    makes the batch row and the child jobs visible together.

    **max_pending:** enforced per chunk with the same per-actor
    partition as :meth:`enqueue_batch`, each chunk's items carry
    their actors' *effective* caps (operator-stored override when
    set, else the ``@actor(...)`` literal, same resolution as
    :meth:`enqueue`), the backend admits every within-cap actor's
    items, and an over-cap actor's items are refused.  On the
    caller-connection path sequential chunks share one transaction,
    so per-chunk admission accumulates to the true aggregate.  On
    the no-connection path each chunk counts against the rows the
    previous committed chunks wrote, which also accumulates to the
    true aggregate, but the refusal surfaces after the committed
    prefix: :class:`~taskq.exceptions.BatchMaxPendingExceededError`
    raises with STREAM-GLOBAL refused item indices and an
    ``admitted_count`` covering every committed item; items after
    the refusing chunk were never attempted.  Retry only the
    refused items (or rely on ``idempotency_key``s).  The atomic
    path (``failure_policy``/``finalizer``, no connection) resolves
    the same effective caps and keeps all-or-nothing: a cap
    violation rolls back the entire single transaction and raises
    plain :class:`~taskq.exceptions.MaxPendingExceededError` with
    nothing committed.

    **unique_for (not applied on batch paths):** actor-declared
    ``unique_for`` is a single-enqueue contract (see
    :meth:`enqueue_batch`'s disclosure), every item writes, items
    sharing an ``(actor, identity_key)`` are not deduplicated, and
    ``unique_for`` items are fully counted toward ``max_pending``.
    This holds on both arms (the per-chunk INSERT and the atomic
    path's chunked INSERT). The ``idempotency_key`` arbiter IS
    applied on both arms: a colliding item returns the existing
    row, exactly as :meth:`enqueue_batch` does.
    """
    if chunk_size < 1 or chunk_size > MAX_BATCH_SIZE:
        raise ValueError(f"chunk_size must be in [1, {MAX_BATCH_SIZE}], got {chunk_size}")

    from taskq._ids import new_job_id

    resolved_batch_id = UUID(bytes=new_job_id().bytes) if batch_id is None else batch_id

    # Peek the iterable, empty raises ValueError.
    it = iter(items)
    try:
        first_item = next(it)
    except StopIteration:
        raise ValueError("items must not be empty") from None

    # Re-chain the first item back into the stream.
    def _chain() -> Iterable[EnqueueItem]:
        yield first_item
        yield from it

    has_batch_extras = failure_policy is not None or finalizer is not None

    # Build finalizer args (without batch_id stamping).
    finalizer_args: EnqueueArgs | None = None
    if finalizer is not None:
        finalizer_args = build_enqueue_args(
            finalizer.actor_ref,
            finalizer.payload,
            scheduled_at=finalizer.scheduled_at,
            priority=finalizer.priority,
            fairness_key=finalizer.fairness_key,
            identity_key=finalizer.identity_key,
            idempotency_key=finalizer.idempotency_key,
            idempotency_scope=finalizer.idempotency_scope,
            metadata=dict(finalizer.metadata),
            start_to_close=finalizer.start_to_close,
            tags=finalizer.tags,
            idempotency_max_bytes=self._idempotency_max_bytes,
            parent_id=current_parent_id(),
        )

    # Build a lazy generator of EnqueueArgs. The payload is validated
    # exactly ONCE per item, inside build_enqueue_args, and a
    # ValidationError from there is re-annotated with the item's index
    # and actor here. Why not validate upfront and discard the result
    # (the previous shape): that ran pydantic-core twice per batch item
    # on a hot path, and the discarded pass could not be reused because
    # build_enqueue_args re-validates internally. Note the idempotency/
    # scheduling checks inside build_enqueue_args now precede payload
    # validation for a doubly-invalid item, the same precedence a
    # single enqueue already has.
    # H4: collect per-item (actor_ref, args_id) as a side effect so handles
    # can be paired by index after the backend returns rows. This avoids
    # using a single actor's result_adapter for all handles (mixed-actor
    # batches would get wrong deserialization).
    item_meta: list[tuple[ActorRef[Any, Any], JobId]] = []

    # The ambient parent context, read ONCE per call (LIB-2): the
    # generator is consumed lazily, possibly mid-transaction, so the
    # contextvar is read at arm entry, never inside the generator.
    ambient_parent_id = current_parent_id()

    # Effective per-actor caps, memoized as actors appear (same
    # resolution as :meth:`enqueue` / :meth:`enqueue_batch`): without
    # it, an operator-stored ``actor_config.max_pending`` on a
    # literal-less actor never reaches the args, the backend's cap
    # groups skip the actor entirely, and the override is silently
    # unenforced. The chunked arm awaits the cache between chunks; the
    # atomic arm pre-warms the snapshot (see below) and peeks it here,
    # synchronously, because the backend consumes this generator
    # mid-transaction where no await is possible.
    effective_mp: dict[str, int | None] = {}

    def _lazy_args(stream: Iterable[EnqueueItem]) -> Iterable[EnqueueArgs]:
        # Why memoized by ACTOR NAME (verdict per new name only): the
        # name -> queue function assumption — enforced at the registry
        # boundary (worker/_bootstrap refuses keys that disagree with
        # ref.name; sync_actor_config CardinalityViolations on
        # duplicates) — so per-name memoization cannot hide a second
        # queue behind one actor.
        for idx, item in enumerate(stream):
            ref = item.actor_ref
            if ref.name not in effective_mp:
                effective_mp[ref.name] = self._capacity_cache.peek_max_pending(
                    ref.name, ref.max_pending
                )
                # Same snapshot verdict as the async arms, warn-once
                # per queue per TTL. Sync by design: the atomic arm's
                # generator runs mid-transaction where no await is
                # possible, and the verdict is a snapshot lookup.
                self._capacity_cache.maybe_warn_unserved_queue(ref.queue, actor=ref.name)
            try:
                args = build_enqueue_args(
                    ref,
                    item.payload,
                    scheduled_at=item.scheduled_at,
                    priority=item.priority,
                    fairness_key=item.fairness_key,
                    identity_key=item.identity_key,
                    idempotency_key=item.idempotency_key,
                    idempotency_scope=item.idempotency_scope,
                    metadata=dict(item.metadata),
                    start_to_close=item.start_to_close,
                    tags=item.tags,
                    idempotency_max_bytes=self._idempotency_max_bytes,
                    # The H5 strip-then-stamp boundary runs inside
                    # build_enqueue_args: any batch_id on item.metadata
                    # is stripped before the library's own is stamped.
                    stamp_batch_id=str(resolved_batch_id),
                    # Stored operator overrides win over the @actor
                    # literal (same resolution as enqueue /
                    # enqueue_batch); peek is synchronous because the
                    # backend consumes this generator mid-transaction
                    # where no await is possible.
                    max_pending=effective_mp[ref.name],
                    parent_id=ambient_parent_id,
                )
            except ValidationError as exc:
                raise _item_payload_error(idx, ref.name, exc) from exc
            item_meta.append((ref, args.id))
            yield args

    # Determine queue from the first item.
    queue = first_item.actor_ref.queue

    # Build BatchRow when failure_policy OR finalizer is set (C3:
    # finalizer-only batches also need a row for list_batches
    # discoverability and finalizer_job_id auto-exclusion in
    # wait_for_batch). When only finalizer is set, failure_threshold=None.
    # expected_size=0 is a sentinel, the backend computes the real count
    # from the iterable (H6: no materialization).
    batch_row: BatchRow | None = None
    if failure_policy is not None or finalizer is not None:
        threshold = failure_policy.failure_threshold if failure_policy is not None else None
        batch_row = BatchRow(
            id=resolved_batch_id,
            queue=queue,
            status="active",
            expected_size=0,
            consecutive_failures=0,
            failure_threshold=threshold,
            finalizer_job_id=finalizer_args.id if finalizer_args is not None else None,
            originating_actor=None,
            created_at=self._clock.now(),
            completed_at=None,
            metadata={},
        )

    all_handles: list[JobHandle[BaseModel | None]] = []
    total_count = 0

    if has_batch_extras and connection is None:
        # Autonomous atomic path. H6: do NOT materialize the iterable ,
        # pass the lazy generator directly to the backend, which consumes
        # it in chunks inside its transaction. expected_size=0 is a
        # sentinel; the backend computes the real count from the items
        # consumed. H4: item_meta is populated as a side effect of the
        # generator being consumed, providing per-item actor_refs and
        # args_ids for handle pairing.
        #
        # Warm the capacity snapshot BEFORE the backend's transaction
        # starts: _lazy_args is a SYNC generator consumed mid-transaction,
        # so it cannot await a (possibly stale-triggering) refresh at
        # yield time. One awaited resolution here, for the first item's
        # actor, whose result the generator also reuses, spends the same
        # single refresh-per-call budget enqueue_batch spends; every
        # actor the stream later introduces resolves from that snapshot
        # via peek_max_pending. A failed refresh fails open to literals,
        # exactly like every other arm.
        first_ref = first_item.actor_ref
        effective_mp[first_ref.name] = await self._capacity_cache.effective_max_pending(
            first_ref.name, first_ref.max_pending
        )
        self._capacity_cache.maybe_warn_unserved_queue(first_ref.queue, actor=first_ref.name)
        if finalizer is not None:
            # The finalizer rides the SAME strict verdict (the
            # per-item verdicts run inside the sync generator, which
            # never sees the finalizer — its args ride the atomic
            # call below): the exact strict bypass the batch arm's
            # fix documents. The snapshot is warm here — the awaited
            # resolution above spent the one refresh — so the verdict
            # is a pure snapshot lookup, zero extra I/O.
            self._capacity_cache.maybe_warn_unserved_queue(
                finalizer.actor_ref.queue, actor=finalizer.actor_ref.name
            )
        all_rows = await self._backend.enqueue_batch_atomic(
            _lazy_args(_chain()),
            batch_id=resolved_batch_id,
            queue=queue,
            batch_row=batch_row,
            finalizer_args=finalizer_args,
            chunk_size=chunk_size,
        )
        non_finalizer_count = len(all_rows) - (1 if finalizer is not None else 0)
        for i in range(non_finalizer_count):
            row = all_rows[i]
            ref, args_id = item_meta[i]
            all_handles.append(
                JobHandle(
                    client=self,
                    row=row,
                    result_adapter=ref.result_adapter,
                    was_existing=(row.id != args_id),
                    _redis_client=self._redis_client,
                    _settings=self._settings,
                )
            )
        total_count = non_finalizer_count
        finalizer_row = all_rows[-1] if finalizer is not None else None
    else:
        # Chunked path (caller-owned connection or no extras). The
        # batch row is created AFTER all chunk inserts: create_batch is
        # INSERT (not upsert), so the row must carry the real
        # expected_size, which is only known once the stream is drained.
        # KNOWN LIMITATION: until the row exists, a child job reaching a
        # terminal state finds no batch row, increment_batch_failures
        # returns (0, None, 0) and the failure is NOT counted toward
        # failure_policy (see the docstring disclosure above). Insert
        # the finalizer first so its returned row id is known for
        # finalizer_job_id (M4).
        stream = _chain()
        finalizer_row = None
        if finalizer is not None:
            assert finalizer_args is not None
            # The finalizer writes FIRST here (M4), BEFORE the chunk
            # loop's first awaited refresh — so this path's strict
            # verdict needs the refresh spent on it explicitly (free
            # when warm, the single-refresh-per-TTL budget; a failed
            # refresh fails the verdict open, the same posture as
            # every other arm). Without the awaited resolution the
            # verdict below would read a possibly-cold snapshot and
            # fail open on a genuinely stranded finalizer.
            await self._capacity_cache.effective_max_pending(
                finalizer.actor_ref.name, finalizer.actor_ref.max_pending
            )
            # The finalizer's strict verdict — the chunk loop's
            # per-item verdicts never see it, and this is the write
            # the caller-connection shape performs.
            self._capacity_cache.maybe_warn_unserved_queue(
                finalizer.actor_ref.queue, actor=finalizer.actor_ref.name
            )
            finalizer_row = await self._backend.enqueue_with_conn(connection, finalizer_args)  # type: ignore[arg-type]  # Why: guarded by has_batch_extras; when connection is provided it is runtime-compatible

        # Consume chunks. The payload is validated exactly ONCE per item
        # (inside build_enqueue_args, via build_batch_args), the previous
        # per-item pre-validation pass ran pydantic-core twice per item and
        # its result was discarded. Every per-item typed error this chunk
        # can raise, pydantic validation from the args build, the jsonb
        # NUL guard and the per-actor cap partition from the backend call
        # , crosses ONE remapping boundary below (_ITEM_ERROR_REMAPS) and
        # leaves annotated at stream-global indices, keeping the
        # index-annotated PayloadValidationError contract (M6) without a
        # second validation on the happy path.
        while True:
            chunk_items = list(islice(stream, chunk_size))
            if not chunk_items:
                break
            # Resolve effective caps for actors NEW to this chunk
            # (memoized in effective_mp across chunks, the TTL'd
            # cache makes this a lookup after the first) before the
            # args are built, so a stored override on a literal-less
            # actor is enforced exactly as enqueue_batch enforces it.
            # Memoizing by actor name assumes name -> queue is a
            # function — enforced at the registry boundary
            # (worker/_bootstrap refuses keys that disagree with
            # ref.name; sync_actor_config CardinalityViolations on
            # duplicates).
            for ci in chunk_items:
                if ci.actor_ref.name not in effective_mp:
                    effective_mp[
                        ci.actor_ref.name
                    ] = await self._capacity_cache.effective_max_pending(
                        ci.actor_ref.name, ci.actor_ref.max_pending
                    )
                    self._capacity_cache.maybe_warn_unserved_queue(
                        ci.actor_ref.queue, actor=ci.actor_ref.name
                    )
            # The stream-global base for every per-item error this
            # chunk can raise, captured BEFORE anything is built or
            # inserted: total_count is exactly the committed prefix
            # (chunks 1..N-1), so a chunk-local item index plus this
            # base names the item's position in the CALLER's stream.
            chunk_offset = total_count
            chunk_args: list[EnqueueArgs] | None = None
            # ONE boundary for every remappable per-item error: the
            # registry's members (and their subclasses, via the
            # isinstance dispatch) re-raise with stream-global item
            # indices; anything else, driver errors, programming
            # errors, crosses untouched.
            try:
                chunk_args = build_batch_args(
                    chunk_items,
                    resolved_batch_id,
                    max_pending_by_actor=effective_mp,
                    parent_id=ambient_parent_id,
                )
                chunk_rows = await self._backend.enqueue_batch(
                    chunk_args, connection=connection
                )  # type: ignore[call-arg]  # Why: asyncpg.Connection is compatible with the protocol's connection parameter at runtime
            except _REMAPPABLE_ITEM_ERROR_TYPES as exc:
                _remap_chunk_item_error(exc, chunk_items, chunk_args, chunk_offset)
            assert (
                chunk_args is not None
            )  # Why: bound inside the try before the backend call; every failure path re-raises above
            for i, row in enumerate(chunk_rows):
                all_handles.append(
                    JobHandle(
                        client=self,
                        row=row,
                        result_adapter=chunk_items[i].actor_ref.result_adapter,
                        was_existing=(row.id != chunk_args[i].id),
                        _redis_client=self._redis_client,
                        _settings=self._settings,
                    )
                )
            total_count += len(chunk_items)

        if batch_row is not None:
            if finalizer_row is not None:
                batch_row = replace(
                    batch_row, finalizer_job_id=finalizer_row.id, expected_size=total_count
                )
            else:
                batch_row = replace(batch_row, expected_size=total_count)
            await self._backend.create_batch(
                batch_row.id,
                batch_row.queue,
                batch_row.expected_size,
                batch_row.failure_threshold,
                batch_row.finalizer_job_id,
                batch_row.originating_actor,
                connection=connection,  # type: ignore[arg-type]  # Why: connection may be None but create_batch handles that
            )

    # Build finalizer handle.
    finalizer_handle: JobHandle[BaseModel | None] | None = None
    if finalizer is not None and finalizer_row is not None:
        finalizer_handle = JobHandle(
            client=self,
            row=finalizer_row,
            result_adapter=finalizer.actor_ref.result_adapter,
            was_existing=(finalizer_row.id != finalizer_args.id)
            if finalizer_args is not None
            else False,
            _redis_client=self._redis_client,
            _settings=self._settings,
        )
        all_handles.append(finalizer_handle)

    logger.debug(
        "batch-streaming-enqueued",
        kind="batch-streaming-enqueued",
        batch_id=str(resolved_batch_id),
        size=total_count,
    )

    return BatchHandle(
        batch_id=resolved_batch_id,
        job_handles=all_handles,
        size=total_count,
        finalizer_handle=finalizer_handle,
    )

get_batch `async` ¶

get_batch(batch_id: UUID) -> BatchRow | None

Fetch a single batch row by ID.

Delegates to :meth:Backend.get_batch. Returns None when the batch does not exist.

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

async def get_batch(self, batch_id: UUID) -> BatchRow | None:
    """Fetch a single batch row by ID.

    Delegates to :meth:`Backend.get_batch`. Returns ``None`` when the
    batch does not exist.
    """
    return await self._backend.get_batch(batch_id)

list_batches `async` ¶

list_batches(filter: BatchFilter) -> list[BatchSummary]

List batches matching filter, returning :class:BatchSummary objects.

Delegates to :meth:Backend.list_batches and maps each (BatchRow, BatchCounts) pair to a :class:BatchSummary with a :class:BatchCompletionStatus derived from the live counts.

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

async def list_batches(
    self,
    filter: BatchFilter,
) -> list[BatchSummary]:
    """List batches matching *filter*, returning :class:`BatchSummary` objects.

    Delegates to :meth:`Backend.list_batches` and maps each
    ``(BatchRow, BatchCounts)`` pair to a :class:`BatchSummary`
    with a :class:`BatchCompletionStatus` derived from the live counts.
    """
    from taskq.batch import BatchCompletionStatus

    pairs = await self._backend.list_batches(filter)
    summaries: list[BatchSummary] = []
    for row, counts in pairs:
        completion = BatchCompletionStatus(
            total=counts.total,
            pending=counts.pending,
            succeeded=counts.succeeded,
            failed=counts.failed,
            cancelled=counts.cancelled,
            crashed=counts.crashed,
            abandoned=counts.abandoned,
        )
        summaries.append(
            BatchSummary(
                batch_id=row.id,
                queue=row.queue,
                status=row.status,
                expected_size=row.expected_size,
                consecutive_failures=row.consecutive_failures,
                failure_threshold=row.failure_threshold,
                finalizer_job_id=row.finalizer_job_id,
                originating_actor=row.originating_actor,
                created_at=row.created_at,
                completed_at=row.completed_at,
                completion=completion,
            )
        )
    return summaries

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 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. max_pending IS enforced, with the same per-actor partition admission and the same effective-cap resolution (operator-stored override when set, else the @actor(...) literal) as :meth:enqueue_batch (see its docstring). 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 :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 :class:~taskq.exceptions.DuplicateIdempotencyKeyError, a same-(idempotency_scope, idempotency_key) pair, whether repeated within the batch or already stored; nothing is written. Callers must pre-deduplicate. One carve-out: during the 01.00.03 pre→post migration window, a key reused across different scopes raises :class:~taskq.exceptions.ScopedIdempotencyMigrationPendingError instead, matching the other enqueue paths.
No unique_for preflight. Actor-declared unique_for is a single-enqueue contract (see :meth:enqueue_batch's disclosure): the COPY writes every item, items sharing an (actor, identity_key) are not deduplicated, and unique_for items are fully counted toward max_pending , same semantics as the unnest batch tier.
max_pending partition admission. One aggregated count runs before the COPY: within-cap actors' rows are written, and an over-cap actor's items are refused, the COPY of the admitted rows commits first, then :class:~taskq.exceptions.BatchMaxPendingExceededError raises naming the refused actors and item indices (retry only those, or rely on idempotency keys).
No JobHandle instances. Only the inserted row count is returned. Use batch_id to query rows post-insert.
All-or-nothing on constraint violations. The entire COPY fails on any constraint violation (duplicate keys, singleton, CHECK), only cap admission partitions.

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 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. ``max_pending`` IS enforced, with the same
    per-actor partition admission and the same effective-cap
    resolution (operator-stored override when set, else the
    ``@actor(...)`` literal) as :meth:`enqueue_batch` (see its
    docstring). 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 :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
      :class:`~taskq.exceptions.DuplicateIdempotencyKeyError`, a
      same-``(idempotency_scope, idempotency_key)`` pair, whether
      repeated within the batch or already stored; nothing is
      written. Callers must pre-deduplicate. One carve-out: during
      the ``01.00.03`` pre→post migration window, a key reused
      across *different* scopes raises
      :class:`~taskq.exceptions.ScopedIdempotencyMigrationPendingError`
      instead, matching the other enqueue paths.
    - **No unique_for preflight.** Actor-declared ``unique_for`` is
      a single-enqueue contract (see :meth:`enqueue_batch`'s
      disclosure): the COPY writes every item, items sharing an
      ``(actor, identity_key)`` are not deduplicated, and
      ``unique_for`` items are fully counted toward ``max_pending``
     , same semantics as the unnest batch tier.
    - **max_pending partition admission.** One aggregated count runs
      before the COPY: within-cap actors' rows are written, and an
      over-cap actor's items are refused, the COPY of the admitted
      rows commits first, then
      :class:`~taskq.exceptions.BatchMaxPendingExceededError` raises
      naming the refused actors and item indices (retry only those,
      or rely on idempotency keys).
    - **No JobHandle instances.** Only the inserted row count is
      returned.  Use ``batch_id`` to query rows post-insert.
    - **All-or-nothing on constraint violations.** The entire COPY
      fails on any constraint violation (duplicate keys, singleton,
      CHECK), only cap admission partitions.

    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 item in items:
        ref = item.actor_ref
        validate_actor_payload(ref.payload_type, item.payload, actor=ref.name)

    # Phase 1.5: Resolve the effective cap per actor, the same
    # resolution and rationale as enqueue_batch (see the comment
    # there): a stored operator override on a literal-less actor must
    # be visible to the backend's cap groups, which skip actors
    # carrying no cap.
    effective_mp: dict[str, int | None] = {}
    for item in items:
        ref = item.actor_ref
        if ref.name not in effective_mp:
            effective_mp[ref.name] = await self._capacity_cache.effective_max_pending(
                ref.name, ref.max_pending
            )
            self._capacity_cache.maybe_warn_unserved_queue(ref.queue, actor=ref.name)

    # Phase 2: Build per-item EnqueueArgs. The ambient parent context
    # is read ONCE per call (LIB-2): every item is a child of the
    # same parent when one is set.
    args_list = build_batch_args(
        items,
        resolved_batch_id,
        max_pending_by_actor=effective_mp,
        parent_id=current_parent_id(),
    )

    # 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 lookup reads the hot jobs table first and falls back to jobs_archive (the same jobs-then-archive probe taskq job show applies), so an id whose row a prune moved to the archive returns that row's handle with :attr:JobHandle.archived True instead of reading as missing; None still means "in neither tier". 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 lookup reads the hot
    ``jobs`` table first and falls back to ``jobs_archive`` (the
    same jobs-then-archive probe ``taskq job show`` applies), so an
    id whose row a prune moved to the archive returns that row's
    handle with :attr:`JobHandle.archived` ``True`` instead of
    reading as missing; ``None`` still means "in neither tier".
    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 _NONE_RESULT_ADAPTER
    )  # type: ignore[assignment]  # Why: TypeAdapter[None] 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,
    )

get_row `async` ¶

get_row(job_id: JobId) -> JobRow | None

Look up a job by id and return the raw :class:JobRow.

Mirrors :meth:get's contract, one backend.get (hot table first, jobs_archive fallback, an archive hit marked archived=True), None when the job exists in neither tier, without the handle machinery or result adapter. For callers that never need a :class:JobHandle, this is the direct form; for the fresh-read case that does want a handle, prefer get plus the handle's row property (still a single round trip).

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

async def get_row(self, job_id: JobId) -> JobRow | None:
    """Look up a job by id and return the raw :class:`JobRow`.

    Mirrors :meth:`get`'s contract, one ``backend.get`` (hot table
    first, ``jobs_archive`` fallback, an archive hit marked
    ``archived=True``), ``None`` when the job exists in neither
    tier, without the handle machinery or
    result adapter. For callers that never need a
    :class:`JobHandle`, this is the direct form; for the fresh-read
    case that does want a handle, prefer ``get`` plus the handle's
    ``row`` property (still a single round trip).
    """
    with self._translate_schema_errors():
        return await self._backend.get(job_id)

backpressure `async` ¶

backpressure(
    queues: Sequence[str], *, parent_id: JobId | None = None
) -> BackpressureSnapshot

Read the submit path's backpressure state for queues.

A read-only typed read (the LIB-2 design, mirroring the reader-method precedent of :meth:get_row / :meth:list) — deliberately NOT an enqueue kwarg: enqueue is generic JobHandle[R] and no flag may flip the reply type.

Per queue (see :class:~taskq.types.QueueBackpressure):

depth — jobs holding a pending slot IN THIS QUEUE (pending + scheduled), the queue-local ops view, ONE indexed aggregate for the whole call (count_pending_jobs_by_queue, the count_active_jobs pattern). The worker-side leader sampler's cache is in-process on the worker and not readable cross-process; the depth here is a real count, not a pretense. REPORTED, not verdict-bearing — the admission cap is per-ACTOR and governs the actor's count across ALL queues, so a queue-local count against an actor cap would false-OK exactly when an actor's traffic splits across queues.
binding_actor / admission_load / effective_max_pending — the VERDICT's basis: the routing actor with the smallest headroom (its OWN pending+scheduled count, ALL queues, from count_pending_jobs — the enqueue_max_pending_count grouping — against its stored cap from the ActorCapacityCache TTL snapshot). The data model has no per-queue depth threshold; the boundary is the actor's.
children_depth — when a fan-out parent is in play (parent_id, or the ambient worker context current_parent_id()), the parent's pending children OUTSIDE the queue, exact via the parent_id ledger stamp (01.00.23_01). REPORTED, not verdict-bearing: every pending child already counts toward its OWN actor's cap (wherever it sits), which is the number admission enforces — the fan-out pressure reaches the verdict through the routing actors' admission loads, and this field is the parent-attributed ledger view of the same rows. Children already in the queue are in its depth and never counted twice.
state — over when any routing actor's OWN count is at or over its cap (the next enqueue for that actor is likely refused with MaxPendingExceededError), ok only when every routing actor resolves and none is over (an unresolvable actor may be over — claiming ok would be the false-ok), unknown when the verdict could not be computed (sick database, the actor-count read missing, no capacity snapshot, a cap that lives only in the @actor literal). reason names the absent half. Fail-open, never a fabricated verdict.

ADVISORY semantics: the race between this check and the later enqueue is acceptable (the signal predicts admission, it does not reserve it); the snapshot carries as_of and cap_age_seconds so its own staleness travels with it (the cap age is bounded by the TTL, default 5s, and reports the last SUCCESS, not the last attempt). Cost: one depth aggregate + one actor-count aggregate per call, plus one more only when a parent is in play; unknown on backend trouble — the read never raises except for a missing schema (:class:SchemaNotMigratedError, the actionable setup defect).

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

async def backpressure(
    self,
    queues: Sequence[str],
    *,
    parent_id: JobId | None = None,
) -> BackpressureSnapshot:
    """Read the submit path's backpressure state for *queues*.

    A read-only typed read (the LIB-2 design, mirroring the
    reader-method precedent of :meth:`get_row` / :meth:`list`) —
    deliberately NOT an ``enqueue`` kwarg: ``enqueue`` is generic
    ``JobHandle[R]`` and no flag may flip the reply type.

    Per queue (see :class:`~taskq.types.QueueBackpressure`):

    * ``depth`` — jobs holding a pending slot IN THIS QUEUE
      (pending + scheduled), the queue-local ops view, ONE indexed
      aggregate for the whole call
      (``count_pending_jobs_by_queue``, the ``count_active_jobs``
      pattern). The worker-side leader sampler's cache is
      in-process on the worker and not readable cross-process; the
      depth here is a real count, not a pretense. REPORTED, not
      verdict-bearing — the admission cap is per-ACTOR and governs
      the actor's count across ALL queues, so a queue-local count
      against an actor cap would false-OK exactly when an actor's
      traffic splits across queues.
    * ``binding_actor`` / ``admission_load`` /
      ``effective_max_pending`` — the VERDICT's basis: the routing
      actor with the smallest headroom (its OWN pending+scheduled
      count, ALL queues, from ``count_pending_jobs`` — the
      ``enqueue_max_pending_count`` grouping — against its stored
      cap from the ``ActorCapacityCache`` TTL snapshot). The data
      model has no per-queue depth threshold; the boundary is the
      actor's.
    * ``children_depth`` — when a fan-out parent is in play
      (*parent_id*, or the ambient worker context
      ``current_parent_id()``), the parent's pending children
      OUTSIDE the queue, exact via the ``parent_id`` ledger stamp
      (01.00.23_01). REPORTED, not verdict-bearing: every pending
      child already counts toward its OWN actor's cap (wherever it
      sits), which is the number admission enforces — the fan-out
      pressure reaches the verdict through the routing actors'
      admission loads, and this field is the parent-attributed
      ledger view of the same rows. Children already in the queue
      are in its ``depth`` and never counted twice.
    * ``state`` — ``over`` when any routing actor's OWN count is at
      or over its cap (the next enqueue for that actor is likely
      refused with ``MaxPendingExceededError``), ``ok`` only when
      every routing actor resolves and none is over (an
      unresolvable actor may be over — claiming ok would be the
      false-ok), ``unknown`` when the verdict could not be computed
      (sick database, the actor-count read missing, no capacity
      snapshot, a cap that lives only in the ``@actor`` literal).
      ``reason`` names the absent half. Fail-open, never a
      fabricated verdict.

    ADVISORY semantics: the race between this check and the later
    enqueue is acceptable (the signal predicts admission, it does
    not reserve it); the snapshot carries ``as_of`` and
    ``cap_age_seconds`` so its own staleness travels with it (the
    cap age is bounded by the TTL, default 5s, and reports the last
    SUCCESS, not the last attempt). Cost: one depth aggregate + one
    actor-count aggregate per call, plus one more only when a
    parent is in play; ``unknown`` on backend trouble — the read
    never raises except for a missing schema
    (:class:`SchemaNotMigratedError`, the actionable setup defect).
    """
    with self._translate_schema_errors():
        await self._capacity_cache.refresh()
        return await read_backpressure(
            self._backend,
            self._capacity_cache,
            list(dict.fromkeys(queues)),
            parent_id=parent_id if parent_id is not None else current_parent_id(),
            read_timeout=DEFAULT_CAPACITY_READ_TIMEOUT,
        )

list `async` ¶

list(filter: JobFilter) -> JobPage

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

filter.status accepts a single :data:JobStatus or a sequence of statuses (e.g. JobFilter(status=["pending", "running"])).

filter.unfinished is a meta-filter: unfinished=True selects non-terminal statuses (pending, scheduled, running, 'not yet finished', not 'currently executing') and unfinished=False selects terminal ones. See :class:JobFilter for full semantics.

next_cursor is returned for every ordering, encoded from the columns that ordering actually sorts by, and is None exactly on the last page: the backend is asked for one row past the limit, so a last page that happens to fill the limit is still known to be the last, and a caller paging until the cursor runs out never fetches an empty trailing page. filter.limit is capped at :data:~taskq.backend._protocol.MAX_JOB_LIST_LIMIT (ValueError above it); at the ceiling there is no room to look past the limit, so a full page there carries a cursor that may lead to one empty page.

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

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

    ``filter.status`` accepts a single :data:`JobStatus` or a
    sequence of statuses (e.g. ``JobFilter(status=["pending",
    "running"])``).

    ``filter.unfinished`` is a meta-filter: ``unfinished=True`` selects
    *non-terminal* statuses (pending, scheduled, running, 'not yet
    finished', not 'currently executing') and ``unfinished=False`` selects
    terminal ones.  See :class:`JobFilter` for full semantics.

    ``next_cursor`` is returned for every ordering, encoded from the
    columns that ordering actually sorts by, and is ``None`` exactly on
    the last page: the backend is asked for one row past the limit, so
    a last page that happens to fill the limit is still known to be the
    last, and a caller paging until the cursor runs out never fetches an
    empty trailing page. ``filter.limit`` is capped at
    :data:`~taskq.backend._protocol.MAX_JOB_LIST_LIMIT` (``ValueError``
    above it); at the ceiling there is no room to look past the limit,
    so a full page there carries a cursor that may lead to one empty
    page.
    """
    if filter.limit > MAX_JOB_LIST_LIMIT:
        raise ValueError(
            f"limit must be <= {MAX_JOB_LIST_LIMIT}, got {filter.limit}; page with cursor "
            "for larger result sets"
        )
    if filter.limit < MAX_JOB_LIST_LIMIT:
        probe = dataclasses.replace(filter, limit=filter.limit + 1)
    else:
        probe = filter
    with self._translate_schema_errors():
        fetched = await self._backend.list_jobs(probe)
    rows = fetched[: filter.limit]
    next_cursor: str | None = None
    if rows and len(fetched) >= probe.limit:
        next_cursor = encode_job_cursor(rows[-1], filter.order_by)
    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 (carrying job_id), matching Python's stdlib idiom for "asked for an entry by id; it isn't there" — note this means a typo'd id and a pruned job are both :class:KeyError, never a "not found" result; check existence with :meth:get first when the distinction matters.

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: the status can move between the two reads (a TOCTOU window inherent to a request-write protocol), so a handler that must not race the transition should re-check new_status rather than assume previous_status.

Metrics: 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` (carrying *job_id*), matching
    Python's stdlib idiom for "asked for an entry by id; it isn't
    there" — note this means a typo'd id and a pruned job are both
    :class:`KeyError`, never a "not found" result; check existence
    with :meth:`get` first when the distinction matters.

    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: the status can move between the two reads (a TOCTOU
    window inherent to a request-write protocol), so a handler that
    must not race the transition should re-check ``new_status``
    rather than assume ``previous_status``.

    Metrics: 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=str(job_id),
        previous_status=previous_status,
        cancellation_initiated=initiated,
    )
    return result

cancel_where `async` ¶

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

Cancel all jobs matching filter in a single set-based operation.

Pending/scheduled jobs are moved straight to terminal 'cancelled' (no running actor to cooperate with). Running jobs get cancel_phase=1 set (cooperative cancel), the worker's heartbeat-driven cancel controller observes the phase change and sets the in-process cancel_event.

Guardrail: a filter with no predicates (no queue, status, actor, identity_key, batch_id, tags, active, or created_before) is rejected with :class:EmptyFilterError unless allow_empty_filter=True is passed.

Filter fields used: queue, status, actor, identity_key, batch_id, tags, active, created_before. The limit, cursor, and order_by fields are ignored.

Returns a :class:BulkCancelResult with counts and affected IDs.

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

async def cancel_where(
    self,
    filter: JobFilter,
    reason: str | None = None,
    *,
    allow_empty_filter: bool = False,
) -> BulkCancelResult:
    """Cancel all jobs matching *filter* in a single set-based operation.

    Pending/scheduled jobs are moved straight to terminal 'cancelled'
    (no running actor to cooperate with). Running jobs get
    ``cancel_phase=1`` set (cooperative cancel), the worker's
    heartbeat-driven cancel controller observes the phase change and
    sets the in-process ``cancel_event``.

    **Guardrail:** a filter with no predicates (no queue, status,
    actor, identity_key, batch_id, tags, active, or created_before)
    is rejected with :class:`EmptyFilterError` unless
    ``allow_empty_filter=True`` is passed.

    **Filter fields used:** ``queue``, ``status``, ``actor``,
    ``identity_key``, ``batch_id``, ``tags``, ``active``,
    ``created_before``. The ``limit``, ``cursor``, and ``order_by``
    fields are ignored.

    Returns a :class:`BulkCancelResult` with counts and affected IDs.
    """
    from taskq.obs import record_cancel_requested

    if not allow_empty_filter and not filter.has_predicates():
        raise EmptyFilterError()

    record_cancel_requested()

    with self._translate_schema_errors():
        return await self._backend.cancel_where(filter, 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,
    owner: Literal["code", "operator"] = "operator",
) -> ScheduleHandle

Create a cron schedule. Raises :class:ValueError if both payload_factory and static_payload are provided, or if cron_expr is invalid.

owner declares who owns the schedule's enable/disable lifecycle. "operator" (the default here: a client-created schedule is an operator act) records operator ownership from birth, so a schedule created disabled carries disabled_by='operator' and no worker boot ever reverts it; "code" matches a @cron declaration, whose stale auto-disable a boot reverts (see :class:CronScheduleSpec).

The (actor, name) UNIQUE constraint means each (actor, name) pair may have at most one schedule; a second create_schedule for the same pair raises asyncpg.UniqueViolationError (PG) or :class:ValueError (in-memory). Pass distinct name values to run several cron schedules per actor (e.g. a per-property sync).

When identity_key is set, the cron loop propagates it to cron-fired jobs so they dedup against on-demand jobs for the same business key.

Does NOT validate actor existence at creation time, any string actor name is accepted (validation is deferred to fire time).

The first next_fire_at is seeded from the clock that arbitrates its due-check: on a Postgres-backed client the PG server clock is read first (one-row SELECT clock_timestamp() via the backend's pool, mirroring the worker bootstrap), so app↔DB clock skew cannot shift the fire chain; on a pool-less (in-memory) client the seed comes from the client's injected Clock. This matters permanently: the cron loop's normal path recomputes every subsequent fire from the STORED fire time (only a miss beyond cron_catch_up_window re-anchors on the server clock), so the seed, not any per-tick correction , fixes the chain's phase for the schedule's life.

Parameters:

Name	Type	Description	Default
`dst_strategy`	`DstStrategy`	How to handle DST gaps and overlaps. `skip` (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,
    owner: Literal["code", "operator"] = "operator",
) -> "ScheduleHandle":
    """Create a cron schedule.  Raises :class:`ValueError` if both
    *payload_factory* and *static_payload* are provided, or if
    *cron_expr* is invalid.

    *owner* declares who owns the schedule's enable/disable lifecycle.
    ``"operator"`` (the default here: a client-created schedule is an
    operator act) records operator ownership from birth, so a schedule
    created disabled carries ``disabled_by='operator'`` and no worker
    boot ever reverts it; ``"code"`` matches a ``@cron`` declaration,
    whose stale auto-disable a boot reverts (see
    :class:`CronScheduleSpec`).

    The ``(actor, name)`` UNIQUE constraint means each ``(actor, name)``
    pair may have at most one schedule; a second ``create_schedule`` for
    the same pair raises ``asyncpg.UniqueViolationError`` (PG) or
    :class:`ValueError` (in-memory).  Pass distinct *name* values to run
    several cron schedules per actor (e.g. a per-property sync).

    When *identity_key* is set, the cron loop propagates it to cron-fired
    jobs so they dedup against on-demand jobs for the same business key.

    Does NOT validate actor existence at creation time, any string
    actor name is accepted (validation is deferred to fire time).

    The first ``next_fire_at`` is seeded from the clock that
    arbitrates its due-check: on a Postgres-backed client the PG
    server clock is read first (one-row ``SELECT clock_timestamp()``
    via the backend's pool, mirroring the worker bootstrap), so
    app↔DB clock skew cannot shift the fire chain; on a pool-less
    (in-memory) client the seed comes from the client's injected
    Clock. This matters permanently: the cron loop's normal path
    recomputes every subsequent fire from the STORED fire time
    (only a miss beyond ``cron_catch_up_window`` re-anchors on the
    server clock), so the seed, not any per-tick correction ,
    fixes the chain's phase for the schedule's life.

    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.
    """
    # Lazy import: croniter (+ dateutil) costs ~16ms at import time and
    # is only needed on the cron-schedule path, not for ``import taskq``.
    from croniter import croniter

    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 = await self._schedule_seed_now()
    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,
        owner=owner,
        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."

When cron_expr changes, the recomputed next_fire_at is seeded from the same clock as create_schedule (the PG server clock on Postgres-backed clients; the client's injected Clock in-memory), the stored chain keeps its server-anchored phase.

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

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

    When *cron_expr* changes, the recomputed ``next_fire_at`` is
    seeded from the same clock as ``create_schedule`` (the PG
    server clock on Postgres-backed clients; the client's injected
    Clock in-memory), the stored chain keeps its server-anchored
    phase.
    """
    # Lazy import: croniter (+ dateutil) costs ~16ms at import time and
    # is only needed on the cron-schedule path, not for ``import taskq``.
    from croniter import croniter

    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 = await self._schedule_seed_now()
        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)

_handle ¶

Generic :class:JobHandle, typed handle to an enqueued job.

Carries a :class:pydantic.TypeAdapter for the actor's return type R, which is the mechanism that prevents R from being a phantom type parameter. The single blocking accessor :meth:wait returns R (never R | None); it raises on missing / failed / timeout.

The handle reads the backend through self._backend for :meth:wait; read-back operations (:meth:status, :meth:refresh, :meth:attempts, :meth:cancel) require a :class:JobsClient and raise :class:RuntimeError when the handle was constructed without one.

all `module-attribute` ¶

__all__ = ['JobHandle']

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._deduplicated_onto_terminal: bool = was_existing and row.status in TERMINAL_STATUSES
    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.

deduplicated_onto_terminal `property` ¶

deduplicated_onto_terminal: bool

This enqueue deduplicated onto a job that had already finished.

was_existing cannot carry this on its own: it is True for every dedup, and the overwhelmingly common dedup, onto a live pending or running job, is the mechanism working, the whole point of an identity key. The case that needs a signal is the rare one where the match was a job that already reached a terminal state: no worker will pick that work up, and with a long dedup horizon the caller can wait for the rest of the window before anyone notices, because the strand looks exactly like the success until the work was needed.

Decided from the row the creating call handed back, so learning this costs no second round trip, and frozen there rather than re-read from the live row, because it is a verdict about the enqueue, not about the job's status now: a dedup onto a running job that later finishes was never a strand. It pairs with the WARN-level enqueue_deduplicated line the backend emits for the same hit: that one is the operator's signal, this one is the caller's.

archived `property` ¶

archived: bool

The last observed row came from the archive tier.

Mirrors :attr:row's last-observed discipline: it reads :attr:row's archived flag, so it is True exactly when the job's row now lives in jobs_archive (the hot row was pruned), and it advances with every :meth:refresh / :meth:status / :meth:wait fetch like the row itself. Constructing from an enqueue always starts False.

actor_name `property` ¶

actor_name: str

The actor this job targets.

queue `property` ¶

queue: str

The queue this job was enqueued on.

row `property` ¶

row: JobRow

The last :class:JobRow this handle observed.

Seeded at construction with the row the creating call fetched, and advanced by every successful row fetch through the handle , :meth:refresh, :meth:status, and :meth:wait's polling loop each record the row they just read. :meth:progress_stream does not advance it, its Redis path fetches no rows, and advancing only on the PG fallback would make the semantics backend-dependent. Reading the property costs no backend round trip: handle = await tq.get(id) followed by handle.row is the single-read pattern for full row state, and a long-lived handle's row stays current as its owner refreshes or waits.

The row is returned by reference (it is frozen and backends hand the handle an isolated row); repeated reads between fetches return the same object.

status `async` ¶

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. Advances :attr:row to the fetched row. An id whose row the prune moved to jobs_archive answers from there (its terminal status, :attr:archived True); only an id that exists in neither tier raises.

Raises:

Type	Description
`RuntimeError`	this handle was constructed without a :class:`JobsClient`.
`KeyError`	the job exists in neither `jobs` nor `jobs_archive`.

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. Advances
    :attr:`row` to the fetched row. An id whose row the prune moved
    to ``jobs_archive`` answers from there (its terminal status,
    :attr:`archived` ``True``); only an id that exists in neither
    tier raises.

    Raises:
        RuntimeError: this handle was constructed without a
            :class:`JobsClient`.
        KeyError: the job exists in neither ``jobs`` nor
            ``jobs_archive``.
    """
    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)
    self._observe(row)
    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. Advances :attr:row to the fetched row, so after a refresh handle.row and the return value are the same row. An id whose row the prune moved to jobs_archive refreshes from there (archived=True on the row); only an id that exists in neither tier raises.

Raises:

Type	Description
`RuntimeError`	this handle was constructed without a :class:`JobsClient`.
`KeyError`	the job exists in neither `jobs` nor `jobs_archive`.

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. Advances :attr:`row` to the
    fetched row, so after a refresh ``handle.row`` and the return
    value are the same row. An id whose row the prune moved to
    ``jobs_archive`` refreshes from there (``archived=True`` on the
    row); only an id that exists in neither tier raises.

    Raises:
        RuntimeError: this handle was constructed without a
            :class:`JobsClient`.
        KeyError: the job exists in neither ``jobs`` nor
            ``jobs_archive``.
    """
    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)
    self._observe(row)
    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`.
`KeyError`	the job does not exist (never enqueued, or pruned past retention); the delegate's contract, a typo'd id and a pruned job are indistinguishable here.

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`.
        KeyError: the job does not exist (never enqueued, or pruned
            past retention); the delegate's contract, a typo'd id and
            a pruned job are indistinguishable here.
    """
    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. Advances :attr:row to each row the polling loop fetches, on return, the terminal row the result was extracted from. A job whose row the prune moved to jobs_archive waits on (and returns) the archived terminal row (:attr:archived True); only an id that exists in neither tier raises KeyError.

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.
`KeyError`	the job exists in neither `jobs` nor `jobs_archive`.
`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. Advances
    :attr:`row` to each row the polling loop fetches, on return,
    the terminal row the result was extracted from. A job whose row
    the prune moved to ``jobs_archive`` waits on (and returns) the
    archived terminal row (:attr:`archived` ``True``); only an id
    that exists in neither tier raises ``KeyError``.

    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.
        KeyError: the job exists in neither ``jobs`` nor
            ``jobs_archive``.
        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._fetch_row_bounded_by(deadline)
        if row is None:
            raise KeyError(self.job_id)
        self._observe(row)
        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 (jittered ±20%) and synthesising events from row diffs; a pool or connection error on a poll is retried on the next one.

Raises :class:NotImplementedError when the backend does not support progress streaming (the in-memory backend sets its supports_progress_stream capability to False, it has no pub/sub), and :class:~taskq.exceptions.StreamUnavailable when the row could not be re-read for 30 s straight — the Postgres poll's own reads, and on the Redis path the safety-net re-reads behind the subscription (an infra error out of a re-read is retried inside a 30 s budget, one warning per failure; a non-infra error propagates untouched).

The first event on the Redis path is the row as it stands when the stream is opened — the same contract :meth:TaskQ.stream documents: an already-terminal job yields its terminal event and returns instead of waiting on a channel that will never speak again, and a non-terminal row carrying content (progress has moved, or the status moved past the row the handle was constructed with) is replayed as the first event, so a fresh consumer — a re-attach after a transport cut — does not starve until the actor's next publish. A row with nothing to replay (progress_seq == 0 and no status change) yields nothing extra.

The stream terminates: every poll timeout re-reads the row as a safety net (a terminal write whose pub/sub message was dropped — Redis pub/sub is fire-and-forget — still ends the stream), and a terminal=True event ends it as on the PG path. The one case that never terminates on the Redis path is a job id that reads back as missing: the stream cannot know the job's state, so it falls through to the subscription (the PG fallback warns and ends instead).

Does not advance :attr:row, the Redis path fetches no rows through the subscription (the snapshot read and the on-timeout safety net below read through _backend directly without recording), and advancing only there would make the semantics backend-dependent.

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 (jittered ±20%) and synthesising events from row
    diffs; a pool or connection error on a poll is retried on the next
    one.

    Raises :class:`NotImplementedError` when the backend does not
    support progress streaming (the in-memory backend sets its
    ``supports_progress_stream`` capability to False, it has no
    pub/sub), and :class:`~taskq.exceptions.StreamUnavailable` when
    the row could not be re-read for 30 s straight — the Postgres
    poll's own reads, and on the Redis path the safety-net re-reads
    behind the subscription (an infra error out of a re-read is
    retried inside a 30 s budget, one warning per failure; a
    non-infra error propagates untouched).

    The first event on the Redis path is the row as it stands when
    the stream is opened — the same contract :meth:`TaskQ.stream`
    documents: an already-terminal job yields its terminal event and
    returns instead of waiting on a channel that will never speak
    again, and a non-terminal row carrying content (progress has
    moved, or the status moved past the row the handle was
    constructed with) is replayed as the first event, so a fresh
    consumer — a re-attach after a transport cut — does not starve
    until the actor's next publish. A row with nothing to replay
    (``progress_seq == 0`` and no status change) yields nothing
    extra.

    The stream terminates: every poll timeout re-reads the row as a
    safety net (a terminal write whose pub/sub message was dropped —
    Redis pub/sub is fire-and-forget — still ends the stream), and a
    ``terminal=True`` event ends it as on the PG path. The one case
    that never terminates on the Redis path is a job id that reads
    back as missing: the stream cannot know the job's state, so it
    falls through to the subscription (the PG fallback warns and ends
    instead).

    Does not advance :attr:`row`, the Redis path fetches no rows
    through the subscription (the snapshot read and the on-timeout
    safety net below read through ``_backend`` directly without
    recording), and advancing only there would make the semantics
    backend-dependent.
    """
    # getattr with the protocol default (True), not a bare attribute
    # read: ``Backend`` is a Protocol, so a structural (duck-typed)
    # backend that predates this capability does not inherit the
    # ClassVar default, and a bare read would turn the pre-capability
    # behavior (the stream runs and fails however its transport fails)
    # into an AttributeError on the flag read itself. The tolerant read
    # keeps missing-flag == True == exactly the pre-capability
    # behavior; the subscribe_* wake capabilities in worker/ use the
    # same pattern for structural backends.
    if not getattr(self._backend, "supports_progress_stream", True):
        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:
        # The snapshot TaskQ.stream's Redis arm takes before
        # subscribing: a job already terminal when this call was made
        # has nothing left to publish, so without this read the
        # generator waited on a silent channel forever (red-proven:
        # tests/test_public_api_ergonomics.py). A missing row falls
        # through to the subscription unchanged.
        current = await self._backend.get(self.job_id)
        if current is not None and current.status in TERMINAL_STATUSES:
            yield self._row_to_progress_event(current, status_changed=True)
            return
        # Native replay for a fresh consumer: a non-terminal row that
        # carries content — progress has moved (progress_seq > 0), or
        # the status moved past the row the handle was constructed
        # with — is yielded as the FIRST event, the same "the row as
        # it stands when the stream is opened" contract TaskQ.stream
        # documents. Without it a re-attach after a transport cut
        # starved until the actor's NEXT publish, minutes on a slow
        # actor. A seq-0 row unchanged from the handle's row yields
        # nothing: the consumer already holds that state, and a
        # phantom first event would double-deliver it. Overlap with
        # the live tail is deduped downstream (decode drops
        # seq <= last_seq below, the on-timeout re-fetch drops
        # unchanged rows).
        if current is not None and (
            current.progress_seq > 0 or current.status != self._row.status
        ):
            # The same discriminator the admission condition above
            # uses: a replay whose status moved past the row the
            # handle holds is a state_change, matching the PG
            # fallback's first event (its last_status seeds None, so
            # its first event is always kind="state_change");
            # progress-only movement stays kind="progress".
            yield self._row_to_progress_event(
                current, status_changed=current.status != self._row.status
            )
        # aclosing: closed FROM UPSTREAM on every exit that is not the
        # transport's own exhaustion - a consumer's ``break`` on the
        # terminal event closes it via GeneratorExit inside the
        # consumer's aclosing, never by draining to StopAsyncIteration.
        # A bare ``async for`` would abandon it suspended at its yield
        # for the GC's asyncgen finalizer to close later, one
        # ``async_generator_athrow`` task per abandoned generator - the
        # residue the loop-leak guard names.
        async with contextlib.aclosing(
            self._progress_stream_redis(
                last_seq=current.progress_seq if current is not None else -1,
                last_status=current.status if current is not None else None,
            )
        ) as feed:
            async for event in feed:
                yield event
    else:
        # aclosing: same upstream-close contract as the Redis path above.
        async with contextlib.aclosing(self._progress_stream_pg()) as feed:
            async for event in feed:
                yield event

ActorsClient ¶

ActorsClient(pool: Pool, *, schema: str = 'taskq')

Pool-wrapping facade for actor configuration operations.

Acquires a connection from the injected pool for each call, delegates to taskq.actor_config_ops, and returns the result. The caller must have opened the pool; this class does not manage its lifecycle.

.. note:: This client is Postgres-only, it delegates to :mod:taskq.actor_config_ops, which executes raw SQL against the actor_config table. The :class:~taskq.backend._protocol.Backend protocol does not include actor config operations, so :class:~taskq.testing.InMemoryBackend does not support ActorsClient.

Parameters¶

pool: An open asyncpg.Pool. The caller retains ownership. schema: TaskQ schema name. Defaults to "taskq".

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

def __init__(self, pool: "asyncpg.Pool", *, schema: str = "taskq") -> None:
    self._pool = pool
    self._schema = schema

list `async` ¶

list() -> list[ActorConfigRow]

List all stored actor_config rows, ordered by actor name.

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

async def list(self) -> list[ActorConfigRow]:
    """List all stored actor_config rows, ordered by actor name."""
    async with self._pool.acquire() as conn:
        return await list_actor_configs(conn, schema=self._schema)

get `async` ¶

get(actor: str) -> ActorConfigRow | None

Get one actor_config row, or None if not found.

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

async def get(self, actor: str) -> ActorConfigRow | None:
    """Get one actor_config row, or ``None`` if not found."""
    async with self._pool.acquire() as conn:
        return await get_actor_config(conn, actor, schema=self._schema)

set_capacity `async` ¶

set_capacity(
    actor: str,
    *,
    max_concurrent: int | Unset | None = UNSET,
    max_pending: int | Unset | None = UNSET,
    result_ttl: float | Unset | None = UNSET,
) -> ActorConfigRow | None

Update capacity fields on an existing actor_config row.

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

async def set_capacity(
    self,
    actor: str,
    *,
    max_concurrent: int | Unset | None = UNSET,
    max_pending: int | Unset | None = UNSET,
    result_ttl: float | Unset | None = UNSET,
) -> ActorConfigRow | None:
    """Update capacity fields on an existing actor_config row."""
    async with self._pool.acquire() as conn:
        return await set_actor_config_capacity(
            conn,
            actor,
            max_concurrent=max_concurrent,
            max_pending=max_pending,
            result_ttl=result_ttl,
            schema=self._schema,
        )

deregister `async` ¶

deregister(
    actor: str,
    *,
    force: bool = False,
    purge_queue: bool = False,
) -> DeregisterResult

Deregister an actor with safety checks.

See :func:taskq.actor_config_ops.deregister_actor for the full semantics.

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

async def deregister(
    self,
    actor: str,
    *,
    force: bool = False,
    purge_queue: bool = False,
) -> DeregisterResult:
    """Deregister an actor with safety checks.

    See :func:`taskq.actor_config_ops.deregister_actor` for
    the full semantics.
    """
    async with self._pool.acquire() as conn:
        return await deregister_actor(
            conn,
            actor,
            force=force,
            purge_queue=purge_queue,
            schema=self._schema,
        )

Client¶

_taskq ¶

__all__ module-attribute ¶

logger module-attribute ¶

EventRow dataclass ¶

event_id instance-attribute ¶

job_id instance-attribute ¶

occurred_at instance-attribute ¶

kind instance-attribute ¶

detail instance-attribute ¶

ActorsClient ¶

Parameters¶

list async ¶

get async ¶

set_capacity async ¶

deregister async ¶

JobEvent ¶

model_config class-attribute instance-attribute ¶

job_id instance-attribute ¶

status instance-attribute ¶

progress_state instance-attribute ¶

progress_seq instance-attribute ¶

terminal instance-attribute ¶

TaskQ ¶

Parameters¶

actors property ¶

backend property ¶

open async ¶

close async ¶

reload_credentials async ¶

__aenter__ async ¶

__aexit__ async ¶

enqueue async ¶

enqueue_batch async ¶

enqueue_batch_streaming async ¶

get_batch async ¶

list_batches async ¶

enqueue_batch_fast async ¶

get async ¶

get_row async ¶

list async ¶

backpressure async ¶

cancel async ¶

cancel_where async ¶

retry_job async ¶

create_schedule async ¶

list_schedules async ¶

update_schedule async ¶

delete_schedule async ¶

stream async ¶

watch_reclaims async ¶

Cursor and duplicate semantics¶

Shutdown and backpressure¶

Transport¶

Raises¶

orjson_response_class ¶

_jobs ¶

__all__ module-attribute ¶

logger module-attribute ¶

JobsClient ¶

backend property ¶

close async ¶

invalidate_actor_capacity_cache ¶

enqueue async ¶

enqueue_batch async ¶

enqueue_batch_streaming async ¶

get_batch async ¶

list_batches async ¶

enqueue_batch_fast async ¶

get async ¶

get_row async ¶

backpressure async ¶

list async ¶

cancel async ¶

cancel_where async ¶

create_schedule async ¶

list_schedules async ¶

update_schedule async ¶

delete_schedule async ¶

_handle ¶

all `module-attribute` ¶

logger `module-attribute` ¶

EventRow `dataclass` ¶

event_id `instance-attribute` ¶

job_id `instance-attribute` ¶

occurred_at `instance-attribute` ¶

kind `instance-attribute` ¶

detail `instance-attribute` ¶

list `async` ¶

get `async` ¶

set_capacity `async` ¶

deregister `async` ¶

model_config `class-attribute` `instance-attribute` ¶

job_id `instance-attribute` ¶

status `instance-attribute` ¶

progress_state `instance-attribute` ¶

progress_seq `instance-attribute` ¶

terminal `instance-attribute` ¶

actors `property` ¶

backend `property` ¶

open `async` ¶

close `async` ¶

reload_credentials `async` ¶

aenter `async` ¶

aexit `async` ¶

enqueue `async` ¶

enqueue_batch `async` ¶

enqueue_batch_streaming `async` ¶

get_batch `async` ¶

list_batches `async` ¶

enqueue_batch_fast `async` ¶

get `async` ¶

get_row `async` ¶

list `async` ¶

backpressure `async` ¶

cancel `async` ¶

cancel_where `async` ¶

retry_job `async` ¶

create_schedule `async` ¶

list_schedules `async` ¶

update_schedule `async` ¶

delete_schedule `async` ¶

stream `async` ¶

watch_reclaims `async` ¶

all `module-attribute` ¶

logger `module-attribute` ¶

backend `property` ¶

close `async` ¶

enqueue `async` ¶

enqueue_batch `async` ¶

enqueue_batch_streaming `async` ¶

get_batch `async` ¶

list_batches `async` ¶

enqueue_batch_fast `async` ¶

get `async` ¶

get_row `async` ¶

backpressure `async` ¶

list `async` ¶

cancel `async` ¶

cancel_where `async` ¶

create_schedule `async` ¶

list_schedules `async` ¶

update_schedule `async` ¶

delete_schedule `async` ¶

all `module-attribute` ¶

was_existing `instance-attribute` ¶

job_id `property` ¶

deduplicated_onto_terminal `property` ¶

archived `property` ¶

actor_name `property` ¶

queue `property` ¶

row `property` ¶

status `async` ¶

refresh `async` ¶

attempts `async` ¶

cancel `async` ¶

wait `async` ¶

progress_stream `async` ¶

list `async` ¶

get `async` ¶