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']

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

async for event in tq.stream(job_id):
    await websocket.send_json(event.model_dump())

model_config class-attribute instance-attribute

model_config = ConfigDict(frozen=True)

job_id instance-attribute

job_id: JobId

status instance-attribute

status: JobStatus

progress_state instance-attribute

progress_state: dict[str, object]

progress_seq instance-attribute

progress_seq: int

terminal instance-attribute

terminal: bool

TaskQ

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

Postgres-backed TaskQ client.

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

Parameters

dsn: Postgres DSN string. Mutually exclusive with pool. pool: An already-open asyncpg.Pool. The caller retains ownership; close() will not close it. pool_factory: A zero-arg async factory returning an asyncpg.Pool - the same :data:~taskq.connections.PoolFactory the worker takes via :class:~taskq.connections.WorkerConnections. TaskQ invokes it at open() and owns the result (close() closes it), and :meth:reload_credentials re-invokes it to rotate the pool in place. This is the client-side equivalent of a <role>_pool_factory: pair it with :func:taskq.auth.make_pg_pool_factory for a rotating-credential deployment. Mutually exclusive with dsn and pool. pg_provider: A :class:~taskq.auth.PgCredentialProvider. Sugar for pool_factory=make_pg_pool_factory(dsn, pg_provider, min_size=min_pool_size, max_size=max_pool_size) - it is exactly that call and nothing more, so the two paths share one mechanism. Requires dsn; use pool_factory directly when you need the factory's other hooks (init, server_settings, command_timeout). schema: TaskQ schema name. Defaults to "taskq". min_pool_size: Minimum pool connections. Only used when dsn is provided. max_pool_size: Maximum pool connections. Only used when dsn is provided. redis_url: Redis URL string. Mutually exclusive with redis_client. The library creates and owns the Redis client; close() will close it. redis_client: An already-open redis.asyncio.Redis client. The caller retains ownership; close() will not close it. Mutually exclusive with redis_url. pg_conn_factory: A zero-arg async factory returning an asyncpg.Connection for the LISTEN/NOTIFY transport used by :meth:stream. Mutually exclusive with listen_conn. Takes precedence over dsn when set. Use this when you have no DSN (e.g. AAD-managed-identity auth) but still want streaming. TaskQ owns and closes the connection produced by the factory per stream() call. listen_conn: A pre-constructed asyncpg.Connection for the LISTEN transport. Caller-owned; TaskQ does not close it. Mutually exclusive with pg_conn_factory. Takes precedence over dsn when set. Use this to share a dedicated LISTEN conn across callers. poll_timeout: Maximum seconds to wait between transport wakeups before re-fetching job state. Defaults to 30.0.

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

def __init__(
    self,
    *,
    dsn: str | None = None,
    pool: "asyncpg.Pool | None" = None,
    pool_factory: "PoolFactory | None" = None,
    pg_provider: "PgCredentialProvider | None" = None,
    schema: str = "taskq",
    min_pool_size: int = 1,
    max_pool_size: int = 5,
    redis_url: str | None = None,
    redis_client: Any | None = None,
    pg_conn_factory: "ConnFactory | None" = None,
    listen_conn: "asyncpg.Connection | None" = None,
    poll_timeout: float = 30.0,
    reclaim_event_visibility_delay: timedelta | None = None,
) -> 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 make_pg_pool_factory

        pool_factory = make_pg_pool_factory(
            dsn, pg_provider, min_size=min_pool_size, max_size=max_pool_size
        )
        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.
    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

actors property

actors: ActorsClient

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

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

open async

open() -> None

Open the connection pool and prepare the client.

Called automatically by __aenter__. Safe to call explicitly for frameworks that manage lifecycle outside an async with block. Raises :class:RuntimeError if already open.

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

async def open(self) -> None:
    """Open the connection pool and prepare the client.

    Called automatically by ``__aenter__``. Safe to call explicitly
    for frameworks that manage lifecycle outside an ``async with`` block.
    Raises :class:`RuntimeError` if already open.
    """
    if self._client is not None:
        raise RuntimeError("TaskQ is already open")

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

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

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

    pool = self._pool
    assert pool is not None
    deps = _ClientDeps(
        settings=_ClientSettings(schema_name=self._schema),
        worker_pool=pool,
        heartbeat_pool=pool,
    )
    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
        ),
    )
    # 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.
    load_data: dict[str, str] = {"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)
    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.

close async

close() -> None

Close the client and release the pool if owned.

Called automatically by __aexit__. Safe to call explicitly. No-op if already closed.

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

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

    Called automatically by ``__aexit__``. Safe to call explicitly.
    No-op if already closed.
    """
    if self._client is not None:
        await self._client.close()
        self._client = None
        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() LISTEN fallback - at the new pool, then closes the old one with the same bounded drain used at close().

Call this on a schedule shorter than your credential's lifetime (an Entra access token is typically ~60 min), or on any signal your issuer gives you. It is not needed for the ordinary token refresh that :func:taskq.auth.make_pg_pool_factory already does per physical connection; it is how you drop sessions opened under a revoked credential, and the only way to pick up a changed username, which asyncpg resolves once per pool.

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

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

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

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()`` LISTEN fallback - at the new pool, then closes the old
    one with the same bounded drain used at ``close()``.

    Call this on a schedule shorter than your credential's lifetime (an
    Entra access token is typically ~60 min), or on any signal your issuer
    gives you. It is not needed for the ordinary token refresh that
    :func:`taskq.auth.make_pg_pool_factory` already does per physical
    connection; it is how you drop sessions opened under a revoked
    credential, and the only way to pick up a **changed username**, which
    asyncpg resolves once per pool.

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

    If the factory fails, the exception propagates and the **current pool
    is left untouched and serving**: a transient token-endpoint outage
    must not turn into a client outage.
    """
    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."
        )

    old_pool = self._pool
    # Built before anything is swapped, so a factory failure leaves the
    # live pool in place — see the docstring.
    new_pool = await self._pool_factory()

    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.

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.

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

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

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

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.active is not Celery's 'active' ('currently executing'); it selects by terminality ('not yet finished'). See :class:JobFilter.

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

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

    Delegates to :meth:`JobsClient.list` — note ``filter.active``
    is not Celery's 'active' ('currently executing'); it selects by
    terminality ('not yet finished').  See :class:`JobFilter`.
    """
    return await self._require_open().list(filter)

cancel async

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

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

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

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

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
    )

create_schedule async

create_schedule(
    actor: str | ActorRef[P, R],
    cron_expr: str,
    *,
    timezone: str = "UTC",
    dst_strategy: DstStrategy = "skip",
    payload_factory: str | None = None,
    static_payload: dict[str, object] | None = None,
    name: str = "",
    identity_key: IdentityKey | None = None,
    enabled: bool = True,
) -> ScheduleHandle

Create a cron schedule. Delegates to :meth:JobsClient.create_schedule.

dst_strategy controls how DST gaps/overlaps are handled; see :meth:JobsClient.create_schedule for the full semantics.

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

async def create_schedule[P: BaseModel, R: BaseModel | None](
    self,
    actor: str | ActorRef[P, R],
    cron_expr: str,
    *,
    timezone: str = "UTC",
    dst_strategy: DstStrategy = "skip",
    payload_factory: str | None = None,
    static_payload: dict[str, object] | None = None,
    name: str = "",
    identity_key: IdentityKey | None = None,
    enabled: bool = True,
) -> ScheduleHandle:
    """Create a cron schedule.  Delegates to :meth:`JobsClient.create_schedule`.

    ``dst_strategy`` controls how DST gaps/overlaps are handled; see
    :meth:`JobsClient.create_schedule` for the full semantics.
    """
    return await self._require_open().create_schedule(
        actor,
        cron_expr,
        timezone=timezone,
        dst_strategy=dst_strategy,
        payload_factory=payload_factory,
        static_payload=static_payload,
        name=name,
        identity_key=identity_key,
        enabled=enabled,
    )

list_schedules async

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

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

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

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

update_schedule async

update_schedule(
    schedule_id: UUID,
    *,
    cron_expr: str | None = None,
    enabled: bool | None = None,
    payload_factory: str | None = None,
    static_payload: dict[str, object] | None = None,
    clear_payload_factory: bool = False,
) -> ScheduleRecord

Update a cron schedule. Delegates to :meth:JobsClient.update_schedule.

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

async def update_schedule(
    self,
    schedule_id: UUID,
    *,
    cron_expr: str | None = None,
    enabled: bool | None = None,
    payload_factory: str | None = None,
    static_payload: dict[str, object] | None = None,
    clear_payload_factory: bool = False,
) -> ScheduleRecord:
    """Update a cron schedule.  Delegates to :meth:`JobsClient.update_schedule`."""
    return await self._require_open().update_schedule(
        schedule_id,
        cron_expr=cron_expr,
        enabled=enabled,
        payload_factory=payload_factory,
        static_payload=static_payload,
        clear_payload_factory=clear_payload_factory,
    )

delete_schedule async

delete_schedule(schedule_id: UUID) -> None

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

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

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

stream async

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

Stream live state changes for a job as :class:JobEvent objects.

    Yields one event per observable state transition (status change or
    progress update), terminating automatically when the job reaches a
    terminal state. The final event always has ``terminal=True``.

    Usage::

        async for event in tq.stream(job_id):
            print(event.status, event.progress_state)
            # loop exits automatically when event.terminal is True

        # Or wire directly into a FastAPI SSE response:
        async def event_generator():
            async for event in tq.stream(job_id):
                yield f"data: {event.model_dump_json()}

"

    Raises
    ------
    RuntimeError
        Called before ``tq.open()`` or outside an ``async with`` block.
    KeyError
        The job does not exist.
    RuntimeError
        PG LISTEN transport requested but ``dsn`` was not provided at
        construction (pool-only mode).

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

async def stream(self, job_id: JobId) -> AsyncIterator[JobEvent]:
    """Stream live state changes for a job as :class:`JobEvent` objects.

    Yields one event per observable state transition (status change or
    progress update), terminating automatically when the job reaches a
    terminal state. The final event always has ``terminal=True``.

    Usage::

        async for event in tq.stream(job_id):
            print(event.status, event.progress_state)
            # loop exits automatically when event.terminal is True

        # Or wire directly into a FastAPI SSE response:
        async def event_generator():
            async for event in tq.stream(job_id):
                yield f"data: {event.model_dump_json()}\n\n"

    Raises
    ------
    RuntimeError
        Called before ``tq.open()`` or outside an ``async with`` block.
    KeyError
        The job does not exist.
    RuntimeError
        PG LISTEN transport requested but ``dsn`` was not provided at
        construction (pool-only mode).
    """
    client = self._require_open()
    row = await client.backend.get(job_id)
    if row is None:
        raise KeyError(job_id)

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

    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(
            self._dsn,
            self._schema,
            job_id,
            client,
            self._poll_timeout,
            last_seq=row.progress_seq,
            last_status=row.status,
            pg_conn_factory=self._pg_conn_factory,
            listen_conn=self._listen_conn,
        )
    )
    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.

Cursor and duplicate semantics

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

Shutdown and backpressure

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

Transport

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

Usage::

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

Raises

RuntimeError Called before tq.open() or outside an async with block.

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

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.

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

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

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

    Usage::

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

    Raises
    ------
    RuntimeError
        Called before ``tq.open()`` or outside an ``async with`` block.
    """
    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

_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._warned_unique_for: set[str] = set()
    self._capacity_cache = ActorCapacityCache(backend, ttl=capacity_cache_ttl)
    # 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,
    scheduled_at: datetime | None = None,
    priority: int | None = None,
    schedule_to_close: datetime | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
    identity_key: IdentityKey | None = None,
    fairness_key: str | None = None,
    idempotency_key: IdempotencyKey | None = None,
    idempotency_scope: str | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    metadata: dict[str, object] | None = None,
    tags: list[str] | None = None,
) -> JobHandle[R]

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

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

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

max_pending:

• When the actor's effective max_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 — every mature job queue that offers one (Oban, River) either gives up the atomic INSERT ... ON CONFLICT for a check-then-insert lock (weaker concurrency guarantee) or buckets time into the key itself (coarser, edge-artifact-prone semantics). If your use case genuinely needs "dedupe for the next hour, not forever," encode the window into the scope yourself (e.g. a time-bucketed scope string) until/unless a TTL parameter ships as a separate feature.

• Rolling-deploy note: if this schema is mid-upgrade (the 01.00.03_01_pre_idempotency_scope.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 best-effort. Concurrent enqueues for the same (actor, identity_key) may both insert; the dispatch CTE's running_identities filter ensures only one runs.

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

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

async def enqueue[P: BaseModel, R: BaseModel | None](
    self,
    ref: ActorRef[P, R],
    payload: P,
    *,
    queue: QueueName | None = None,
    scheduled_at: datetime | None = None,
    priority: int | None = None,
    schedule_to_close: datetime | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
    identity_key: IdentityKey | None = None,
    fairness_key: str | None = None,
    idempotency_key: IdempotencyKey | None = None,
    idempotency_scope: str | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    metadata: dict[str, object] | None = None,
    tags: list[str] | None = None,
) -> JobHandle[R]:
    """Enqueue a job for the given actor and return a typed handle.

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

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

    **max_pending:**

    - When the actor's effective ``max_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 — every mature
      job queue that offers one (Oban, River) either gives up the
      atomic ``INSERT ... ON CONFLICT`` for a check-then-insert lock
      (weaker concurrency guarantee) or buckets time into the key
      itself (coarser, edge-artifact-prone semantics). If your use
      case genuinely needs "dedupe for the next hour, not forever,"
      encode the window into the scope yourself (e.g. a
      time-bucketed scope string) until/unless a TTL parameter ships
      as a separate feature.

    - Rolling-deploy note: if this schema is mid-upgrade (the
      ``01.00.03_01_pre_idempotency_scope.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 **best-effort**. Concurrent
      enqueues for the same ``(actor, identity_key)`` may both insert;
      the dispatch CTE's ``running_identities`` filter ensures only one
      runs.

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

    with enqueue_span(ref.name, resolved_queue, identity_key=identity_key_str) as (
        span,
        extracted_trace_id,
        extracted_span_id,
    ):
        effective_max_pending = await self._capacity_cache.effective_max_pending(
            ref.name, ref.max_pending
        )
        # 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". Both were previously
        # accepted and then dropped in favour of the extracted values,
        # so cross-service propagation silently produced 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,
            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,
        )
        span.set_attribute("messaging.message.id", str(args.id))
        if ref.unique_for is not None and args.identity_key is None:
            self._maybe_warn_unique_for_no_identity(ref)
        with self._translate_schema_errors():
            row = await self._backend.enqueue(args)

    if row.id == args.id:
        logger.debug(
            "job_enqueued",
            kind="job_enqueued",
            job_id=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:

One aggregated SELECT actor, count(*) … WHERE actor = ANY($1) GROUP BY actor is issued for the entire batch. Per-actor effective limits (operator-owned stored value when set, else the @actor(...) literal — same resolution as :meth:enqueue) are checked before the INSERT; any violation raises :class:~taskq.exceptions.MaxPendingExceededError.

idempotency_key collisions:

Items whose idempotency_key collides with an existing row return the existing :class:~taskq.client.JobHandle (same semantics as single-item :meth:enqueue).

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

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:**

    One aggregated ``SELECT actor, count(*) … WHERE actor = ANY($1)
    GROUP BY actor`` is issued for the entire batch.  Per-actor
    effective limits (operator-owned stored value when set, else the
    ``@actor(...)`` literal — same resolution as :meth:`enqueue`)
    are checked before the INSERT; any violation raises
    :class:`~taskq.exceptions.MaxPendingExceededError`.

    **idempotency_key collisions:**

    Items whose ``idempotency_key`` collides with an existing row
    return the existing :class:`~taskq.client.JobHandle` (same
    semantics as single-item :meth:`enqueue`).
    """
    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}",
        )

    # Phase 2: Aggregated max_pending check (one query for the whole batch)
    # Resolve the effective limit per actor (stored value wins over the
    # @actor literal — same resolution as enqueue), then count.
    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
            )
    actors_with_limit = {name: mp for name, mp in effective_mp.items() if mp is not None}

    if actors_with_limit:
        # One aggregated query for all actors that declare max_pending.
        existing_counts = await self._backend.count_pending_jobs(list(actors_with_limit.keys()))
        for actor_name, limit in actors_with_limit.items():
            batch_count = sum(1 for it in items if it.actor_ref.name == actor_name)
            existing_pending_count = existing_counts.get(actor_name, 0)
            # M1: use > (not >=) so a batch that fills the queue exactly
            # to the limit is admitted, matching single-enqueue semantics
            # where current_count >= max_pending rejects (i.e. +1 > limit).
            if existing_pending_count + batch_count > limit:
                from taskq.exceptions import MaxPendingExceededError

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

    # Phase 3: Build per-item EnqueueArgs — carrying the resolved
    # limits so a per-item backend check enforces the same value the
    # aggregated check just admitted.
    args_list = build_batch_args(items, resolved_batch_id, max_pending_by_actor=effective_mp)

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

    # 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 on the caller-owned connection, and the batch row + finalizer are created as the last statements.

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

max_pending: NOT enforced on this path — unlike :meth:enqueue_batch, which runs one aggregated per-actor check before the INSERT, neither the chunked :meth:Backend.enqueue_batch inserts nor the atomic delegation consult max_pending. The caller is responsible for ensuring the stream will not exceed actor limits (the same bulk-import semantics :meth:enqueue_batch_fast discloses).

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

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` on the caller-owned connection,
    and the batch row + finalizer are created as the last
    statements.

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

    **max_pending:** NOT enforced on this path — unlike
    :meth:`enqueue_batch`, which runs one aggregated per-actor
    check before the INSERT, neither the chunked
    :meth:`Backend.enqueue_batch` inserts nor the atomic delegation
    consult ``max_pending``. The caller is responsible for ensuring
    the stream will not exceed actor limits (the same bulk-import
    semantics :meth:`enqueue_batch_fast` discloses).
    """
    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,
        )

    # Build a lazy generator of EnqueueArgs, validating payloads on the fly.
    # 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]] = []

    def _lazy_args(stream: Iterable[EnqueueItem]) -> Iterable[EnqueueArgs]:
        for idx, item in enumerate(stream):
            ref = item.actor_ref
            try:
                ref.payload_type.model_validate(item.payload)
            except ValidationError as exc:
                errs: list[dict[str, object]] = exc.errors()  # type: ignore[assignment]  # Why: pydantic v2 ErrorDetails is a TypedDict (subtype of dict[str, Any]); assignment to list[dict[str,object]] is safe at runtime but pyright cannot prove covariance
                raise PayloadValidationError(
                    f"Payload validation failed for item {idx} (actor={ref.name!r}): {exc}",
                    actor=ref.name,
                    validation_errors=errs,
                ) from exc
            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,
            )
            # Stamp batch_id AFTER build_enqueue_args, which strips any
            # caller-supplied batch_id as a security boundary (H5).
            args = replace(
                args,
                metadata={**args.metadata, "batch_id": str(resolved_batch_id)},
            )
            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.
        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
            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, validating payloads with global index (M6).
        global_idx = 0
        while True:
            chunk_items = list(islice(stream, chunk_size))
            if not chunk_items:
                break
            for ci in chunk_items:
                ref = ci.actor_ref
                try:
                    ref.payload_type.model_validate(ci.payload)
                except ValidationError as exc:
                    errs_v: list[dict[str, object]] = exc.errors()  # type: ignore[assignment]  # Why: pydantic v2 ErrorDetails is a TypedDict; safe at runtime
                    raise PayloadValidationError(
                        f"Payload validation failed for item {global_idx} "
                        f"(actor={ref.name!r}): {exc}",
                        actor=ref.name,
                        validation_errors=errs_v,
                    ) from exc
                global_idx += 1
            chunk_args = build_batch_args(chunk_items, resolved_batch_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
            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 enforce max_pending, does NOT detect or reject idempotency-key collisions (a duplicate key aborts the whole batch instead of being treated as "already enqueued"), and returns a bare row count, not per-job handles — there is no way to await, cancel, or otherwise reference an individual job from the return value. Use :meth:enqueue_batch unless you specifically need COPY-level throughput for a one-shot bulk import/backfill and have already accounted for these gaps.

Returns the count of inserted rows — no :class:~taskq.batch.BatchHandle, no per-job :class:~taskq.client.JobHandle instances.

Validation rules:

• len(items) == 0 raises :class:ValueError.
• len(items) > 50_000 raises :class:ValueError.
• ALL payloads are validated before any INSERT — a single failure raises :class:~taskq.exceptions.PayloadValidationError.

Tradeoffs vs enqueue_batch:

• No idempotency-key collision handling. A duplicate key aborts the entire batch with asyncpg.UniqueViolationError. Callers must pre-deduplicate. 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 max_pending check. The caller is responsible for ensuring the batch won't exceed actor limits.
• No JobHandle instances. Only the inserted row count is returned. Use batch_id to query rows post-insert.
• All-or-nothing atomicity. No partial success — the entire COPY fails on any constraint violation.

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

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

async def enqueue_batch_fast(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | None" = None,
) -> int:
    """Enqueue jobs via COPY FROM protocol for maximum throughput.

    **WARNING — bulk-import semantics, not general-purpose enqueue:**
    this method does NOT enforce ``max_pending``, does NOT detect or
    reject idempotency-key collisions (a duplicate key aborts the
    whole batch instead of being treated as "already enqueued"), and
    returns a bare row **count**, not per-job handles — there is no
    way to await, cancel, or otherwise reference an individual job
    from the return value. Use :meth:`enqueue_batch` unless you
    specifically need COPY-level throughput for a one-shot bulk
    import/backfill and have already accounted for these gaps.

    Returns the count of inserted rows — no :class:`~taskq.batch.BatchHandle`,
    no per-job :class:`~taskq.client.JobHandle` instances.

    **Validation rules:**

    - ``len(items) == 0`` raises :class:`ValueError`.
    - ``len(items) > 50_000`` raises :class:`ValueError`.
    - ALL payloads are validated before any INSERT — a single failure
      raises :class:`~taskq.exceptions.PayloadValidationError`.

    **Tradeoffs vs enqueue_batch:**

    - **No idempotency-key collision handling.** A duplicate key
      aborts the entire batch with ``asyncpg.UniqueViolationError``.
      Callers must pre-deduplicate. 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 max_pending check.** The caller is responsible for
      ensuring the batch won't exceed actor limits.
    - **No JobHandle instances.** Only the inserted row count is
      returned.  Use ``batch_id`` to query rows post-insert.
    - **All-or-nothing atomicity.** No partial success — the entire
      COPY fails on any constraint violation.

    Use for bulk import / backfill with 1K-50K rows where throughput
    matters more than idempotency guarantees.
    """
    from taskq._ids import new_job_id

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

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

    # Phase 1: Validate ALL payloads before any I/O
    for item in items:
        ref = item.actor_ref
        validate_actor_payload(ref.payload_type, item.payload, actor=ref.name)

    # Phase 2: Build per-item EnqueueArgs
    args_list = build_batch_args(items, resolved_batch_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 caller may supply result_adapter because lookups by id do not carry actor identity — typical sources are my_actor.result_adapter (when reuniting with an actor) or TypeAdapter(type(None)) (when only row metadata is needed). When result_adapter is None it defaults to TypeAdapter(type(None)), which is suitable for status-only lookups.

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

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

    Returns ``None`` when the job does not exist; otherwise wraps
    the row in a :class:`JobHandle[R]`. The caller may supply
    ``result_adapter`` because lookups by id do not carry actor
    identity — typical sources are
    ``my_actor.result_adapter`` (when reuniting with an actor) or
    ``TypeAdapter(type(None))`` (when only row metadata is needed).
    When *result_adapter* is ``None`` it defaults to
    ``TypeAdapter(type(None))``, which is suitable for status-only
    lookups.
    """
    adapter: TypeAdapter[R] = (
        result_adapter if result_adapter is not None else TypeAdapter(type(None))
    )  # type: ignore[assignment]  # Why: TypeAdapter(type(None)) returns TypeAdapter[None], which does not narrow to TypeAdapter[R] under pyright; runtime behaviour is correct because None is assignable to the R bound
    with self._translate_schema_errors():
        row = await self._backend.get(job_id)
    if row is None:
        return None
    return JobHandle(
        client=self,
        row=row,
        result_adapter=adapter,
        was_existing=False,
        _redis_client=self._redis_client,
        _settings=self._settings,
    )

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, None when the job does not exist — without the handle machinery or result adapter. For callers that never need a :class:JobHandle, this is the direct form; for the fresh-read case that does want a handle, prefer get plus the handle's row property (still a single round trip).

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

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``, ``None``
    when the job does not exist — without the handle machinery or
    result adapter. For callers that never need a
    :class:`JobHandle`, this is the direct form; for the fresh-read
    case that does want a handle, prefer ``get`` plus the handle's
    ``row`` property (still a single round trip).
    """
    with self._translate_schema_errors():
        return await self._backend.get(job_id)

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.active is a meta-filter — not Celery's 'active': active=True selects non-terminal statuses (pending, scheduled, running — 'not yet finished', not 'currently executing') and active=False selects terminal ones. See :class:JobFilter for full semantics.

next_cursor is returned for every ordering, encoded from the columns that ordering actually sorts by, and is only None on the last page.

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

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.active`` is a meta-filter — **not Celery's 'active'**:
    ``active=True`` selects *non-terminal* statuses (pending,
    scheduled, running — 'not yet finished', not 'currently
    executing') and ``active=False`` selects terminal ones.  See
    :class:`JobFilter` for full semantics.

    ``next_cursor`` is returned for every ordering, encoded from the
    columns that ordering actually sorts by, and is only ``None`` on
    the last page.
    """
    with self._translate_schema_errors():
        rows = await self._backend.list_jobs(filter)
    next_cursor: str | None = None
    if rows and len(rows) == filter.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 — matching Python's stdlib idiom for "asked for an entry by id; it isn't there".

Then calls :meth:Backend.write_cancel_request and reads the row again to capture the new status. The previous_status reflects the row at the first read, not atomically at write-time (TOCTOU per ).

: increments taskq.cancellation.requested exactly once per call, regardless of cancellation_initiated outcome.

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

async def cancel(
    self,
    job_id: JobId,
    reason: str | None = None,
) -> CancelResult:
    """Request cancellation of a job and return a :class:`CancelResult`.

    Reads the row first via :meth:`Backend.get`. If the job does not
    exist, raises :class:`KeyError` — matching Python's stdlib
    idiom for "asked for an entry by id; it isn't there".

    Then calls :meth:`Backend.write_cancel_request` and reads the
    row again to capture the new status. The ``previous_status``
    reflects the row at the first read, not atomically at
    write-time (TOCTOU per  ).

    : increments ``taskq.cancellation.requested`` exactly once
    per call, regardless of ``cancellation_initiated`` outcome.
    """
    from taskq.obs import record_cancel_requested

    record_cancel_requested()

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

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

    result = CancelResult(
        job_id=job_id,
        previous_status=previous_status,
        new_status=new_status,
        cancellation_initiated=initiated,
    )
    logger.debug(
        "cancel_requested",
        kind="cancel_requested",
        job_id=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, or active) is rejected with :class:EmptyFilterError unless allow_empty_filter=True is passed.

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

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

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

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, or active) is rejected with
    :class:`EmptyFilterError` unless ``allow_empty_filter=True`` is
    passed.

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

    Returns a :class:`BulkCancelResult` with counts and affected IDs.
    """
    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,
) -> ScheduleHandle

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

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

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

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

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

Parameters:

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

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

async def create_schedule[P: BaseModel, R: BaseModel | None](
    self,
    actor: str | ActorRef[P, R],
    cron_expr: str,
    *,
    timezone: str = "UTC",
    dst_strategy: DstStrategy = "skip",
    payload_factory: str | None = None,
    static_payload: dict[str, object] | None = None,
    name: str = "",
    identity_key: IdentityKey | None = None,
    enabled: bool = True,
) -> "ScheduleHandle":
    """Create a cron schedule.  Raises :class:`ValueError` if both
    *payload_factory* and *static_payload* are provided, or if
    *cron_expr* is invalid.

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

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

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

    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.
    """
    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,
        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.
    """
    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._client = client
    self._backend: Backend = backend if backend is not None else client.backend  # pyright: ignore[reportOptionalMemberAccess]  # Why: client is guaranteed non-None when backend is None; the ValueError above ensures at least one is provided
    self._redis_client: "redis_async.Redis | None" = _redis_client  # noqa: UP037  # Why: redis_async is under TYPE_CHECKING; string annotation prevents a runtime import cycle.
    self._handle_settings: "TaskQSettings | None" = _settings  # noqa: UP037  # Why: TaskQSettings is under TYPE_CHECKING; string annotation prevents a runtime import cycle.

was_existing instance-attribute

was_existing: bool = was_existing

job_id property

job_id: JobId

The job's unique id.

actor_name property

actor_name: str

The actor this job targets.

queue property

queue: str

The queue this job was enqueued on.

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.

Raises:

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

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

async def status(self) -> JobStatus:
    """Return the current status of this job (live read).

    Cheap, non-blocking: a single ``backend.get`` and a status
    projection. No polling. Use this when you want to know the
    state without waiting for a terminal transition. Advances
    :attr:`row` to the fetched row.

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

Raises:

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

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

async def refresh(self) -> JobRow:
    """Re-read the row from the backend and return the raw
    :class:`JobRow`.

    Useful for callers that want full row state (timestamps,
    attempt counts, error metadata) without going through
    :meth:`wait`. Does not block on terminal state — returns the
    current row whatever its status. Advances :attr:`row` to the
    fetched row, so after a refresh ``handle.row`` and the return
    value are the same row.

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

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

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

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

wait async

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

Block until the job reaches a terminal status, then return R.

Returns the actor's return value, validated through :attr:result_adapter. The result type is R exactly — never R | None. Missing or failed results raise. Advances :attr:row to each row the polling loop fetches — on return, the terminal row the result was extracted from.

Raises:

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

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

async def wait(self, *, timeout: float | None = None) -> R:  # noqa: ASYNC109  # Why: timeout is part of the public API contract; asyncio.timeout() context-manager doesn't fit a polling loop
    """Block until the job reaches a terminal status, then return ``R``.

    Returns the actor's return value, validated through
    :attr:`result_adapter`. The result type is ``R`` exactly —
    never ``R | None``. Missing or failed results raise. Advances
    :attr:`row` to each row the polling loop fetches — on return,
    the terminal row the result was extracted from.

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

    while True:
        row = await self._backend.get(self.job_id)
        if row is None:
            raise KeyError(self.job_id)
        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 and synthesising events from row diffs.

Raises :class:NotImplementedError when the in-memory backend is detected — the in-memory backend does not support pub/sub.

Does not advance :attr:row — the Redis path fetches no rows, and advancing only on the PG fallback would make the semantics backend-dependent.

Yields events until a terminal=True event is produced.

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

async def progress_stream(self) -> AsyncIterator[ProgressEvent]:
    """Stream live progress events for this job.

    When Redis is configured, subscribes to the per-job Redis pub/sub
    channel and yields :class:`~taskq.progress.ProgressEvent` objects in
    real time. When Redis is not available, falls back to polling Postgres
    at 500 ms intervals and synthesising events from row diffs.

    Raises :class:`NotImplementedError` when the in-memory backend is
    detected — the in-memory backend does not support pub/sub.

    Does not advance :attr:`row` — the Redis path fetches no rows,
    and advancing only on the PG fallback would make the semantics
    backend-dependent.

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

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

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

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