Client

TaskQ, JobsClient, JobHandle, and CancelResult.

_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__ = ['JobEvent', 'TaskQ']

logger module-attribute

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

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,
    schema: str = "taskq",
    min_pool_size: int = 1,
    max_pool_size: int = 5,
    redis_url: str | None = None,
    redis_client: Any | None = None,
    poll_timeout: float = 30.0,
)

Postgres-backed TaskQ client.

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

Parameters

dsn: Postgres DSN string. Mutually exclusive with pool. pool: An already-open asyncpg.Pool. The caller retains ownership; close() will not close it. schema: TaskQ schema name. Defaults to "taskq". min_pool_size: Minimum pool connections. Only used when dsn is provided. max_pool_size: Maximum pool connections. Only used when dsn is provided. redis_url: Redis URL string. Mutually exclusive with redis_client. The library creates and owns the Redis client; close() will close it. redis_client: An already-open redis.asyncio.Redis client. The caller retains ownership; close() will not close it. Mutually exclusive with redis_url. poll_timeout: Maximum seconds to wait between transport wakeups before re-fetching job state. Defaults to 30.0.

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

def __init__(
    self,
    *,
    dsn: str | None = None,
    pool: "asyncpg.Pool | None" = None,
    schema: str = "taskq",
    min_pool_size: int = 1,
    max_pool_size: int = 5,
    redis_url: str | None = None,
    redis_client: Any | None = None,
    poll_timeout: float = 30.0,
) -> None:
    if dsn is None and pool is None:
        raise ValueError("TaskQ requires either 'dsn' or 'pool'")
    if dsn is not None and pool is not None:
        raise ValueError("TaskQ accepts 'dsn' or 'pool', not both")
    if redis_url is not None and redis_client is not None:
        raise ValueError("TaskQ accepts 'redis_url' or 'redis_client', not both")

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

open async

open() -> None

Open the connection pool and prepare the client.

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

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

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

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

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

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

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

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

close async

close() -> None

Close the client and release the pool if owned.

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

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

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

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

__aenter__ async

__aenter__() -> TaskQ

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

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

__aexit__ async

__aexit__(*_: object) -> None

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

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

enqueue async

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

Enqueue a job and return a typed handle.

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

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

enqueue_batch async

enqueue_batch(
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: Connection | None = None,
) -> BatchHandle

Enqueue multiple jobs in a single batched INSERT.

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

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

async def enqueue_batch(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | None" = None,
) -> BatchHandle:
    """Enqueue multiple jobs in a single batched INSERT.

    Delegates to :meth:`JobsClient.enqueue_batch`; see its docstring
    for validation rules and idempotency-key collision semantics.
    """
    return await self._require_open().enqueue_batch(
        items,
        batch_id=batch_id,
        connection=connection,
    )

enqueue_batch_fast async

enqueue_batch_fast(
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: Connection | None = None,
) -> int

Enqueue jobs via COPY FROM protocol for maximum throughput.

Delegates to :meth:JobsClient.enqueue_batch_fast; see its docstring for tradeoffs vs the regular :meth:enqueue_batch.

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

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

    Delegates to :meth:`JobsClient.enqueue_batch_fast`; see its
    docstring for tradeoffs vs the regular :meth:`enqueue_batch`.
    """
    return await self._require_open().enqueue_batch_fast(
        items,
        batch_id=batch_id,
        connection=connection,
    )

get async

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

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

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

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

list async

list(filter: JobFilter) -> JobPage

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

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

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

cancel async

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

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

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

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

create_schedule async

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

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

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

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

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

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

list_schedules async

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

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

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

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

update_schedule async

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

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

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

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

delete_schedule async

delete_schedule(schedule_id: UUID) -> None

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

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

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

stream async

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

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

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

    Usage::

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

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

"

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

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

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

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

    Usage::

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

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

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

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

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

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

Public API for job operations.

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

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

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

backend property

backend: Backend

The underlying :class:Backend this client delegates to.

Exposed so :class:JobHandle can read the backend through the client without accessing the private _backend attribute.

close async

close() -> None

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

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

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

enqueue async

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

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

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

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

max_pending:

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

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

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

idempotency_key:

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

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

unique_for:

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

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

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

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

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

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

    **max_pending:**

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

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

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

    **idempotency_key:**

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

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

    **unique_for:**

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

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

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

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

enqueue_batch async

enqueue_batch(
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: Connection | None = None,
) -> BatchHandle

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

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

Validation rules:

• len(items) == 0 raises :class:ValueError.
• len(items) > 1000 raises :class:ValueError.
• ALL payloads are validated before any INSERT. A single failure raises :class:~taskq.exceptions.PayloadValidationError and leaves no rows inserted.

max_pending:

One aggregated SELECT actor, count(*) … WHERE actor = ANY($1) GROUP BY actor is issued for the entire batch. Per-actor limits are checked before the INSERT; any violation raises :class:~taskq.exceptions.MaxPendingExceededError.

idempotency_key collisions:

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

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

async def enqueue_batch(
    self,
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: "asyncpg.Connection | None" = None,
) -> BatchHandle:
    """Enqueue multiple jobs in a single batched INSERT and return a
    :class:`~taskq.batch.BatchHandle`.

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

    **Validation rules:**

    - ``len(items) == 0`` raises :class:`ValueError`.
    - ``len(items) > 1000`` raises :class:`ValueError`.
    - ALL payloads are validated before any INSERT.  A single failure
      raises :class:`~taskq.exceptions.PayloadValidationError` and
      leaves no rows inserted.

    **max_pending:**

    One aggregated ``SELECT actor, count(*) … WHERE actor = ANY($1)
    GROUP BY actor`` is issued for the entire batch.  Per-actor limits
    are checked before the INSERT; any violation raises
    :class:`~taskq.exceptions.MaxPendingExceededError`.

    **idempotency_key collisions:**

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

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

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

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

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

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

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

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

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

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

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

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

enqueue_batch_fast async

enqueue_batch_fast(
    items: list[EnqueueItem],
    *,
    batch_id: UUID | None = None,
    connection: Connection | None = None,
) -> int

Enqueue jobs via COPY FROM protocol for maximum throughput.

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

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

Validation rules:

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

Tradeoffs vs enqueue_batch:

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

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

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

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

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

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

    **Validation rules:**

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

    **Tradeoffs vs enqueue_batch:**

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

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

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

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

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

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

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

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

    return count

get async

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

Look up a job by id.

Returns None when the job does not exist; otherwise wraps the row in a :class:JobHandle[R]. The caller may supply result_adapter because lookups by id do not carry actor identity — typical sources are my_actor.result_adapter (when reuniting with an actor) or TypeAdapter(type(None)) (when only row metadata is needed). When result_adapter is None it defaults to TypeAdapter(type(None)), which is suitable for status-only lookups.

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

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

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

list async

list(filter: JobFilter) -> JobPage

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

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

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

cancel async

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

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

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

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

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

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

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

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

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

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

    record_cancel_requested()

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

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

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

create_schedule async

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

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

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

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

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

Parameters:

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

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

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

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

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

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

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

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

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

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

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

list_schedules async

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

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

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

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

update_schedule async

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

Update a cron schedule. Setting enabled=True clears last_fire_error and resets consecutive_failures to 0.

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

To explicitly clear payload_factory (set the column to NULL), pass clear_payload_factory=True — None for payload_factory means "don't change this field."

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

async def update_schedule(
    self,
    schedule_id: UUID,
    *,
    cron_expr: str | None = None,
    enabled: bool | None = None,
    payload_factory: str | None = None,
    static_payload: dict[str, object] | None = None,
    clear_payload_factory: bool = False,
) -> "ScheduleRecord":
    """Update a cron schedule.  Setting ``enabled=True`` clears
    ``last_fire_error`` and resets ``consecutive_failures`` to 0.

    Raises :class:`ValueError` if both *payload_factory* and
    *static_payload* are provided, or if *cron_expr* is invalid.

    To explicitly clear ``payload_factory`` (set the column to NULL),
    pass ``clear_payload_factory=True`` — ``None`` for payload_factory
    means "don't change this field."
    """
    from taskq.cron import compute_next_fire_after

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

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

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

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

delete_schedule async

delete_schedule(schedule_id: UUID) -> None

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

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

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

_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:attempts) 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.

status async

status() -> JobStatus

Return the current status of this job (live read).

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

Raises:

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

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

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

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

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

refresh async

refresh() -> JobRow

Re-read the row from the backend and return the raw :class:JobRow.

Useful for callers that want full row state (timestamps, attempt counts, error metadata) without going through :meth:wait. Does not block on terminal state — returns the current row whatever its status.

Raises:

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

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

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

    Useful for callers that want full row state (timestamps,
    attempt counts, error metadata) without going through
    :meth:`wait`. Does not block on terminal state — returns the
    current row whatever its status.

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

attempts async

attempts() -> list[AttemptRow]

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

Raises:

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

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

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

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

cancel async

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

Delegate to :meth:JobsClient.cancel.

Raises:

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

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

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

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

wait async

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

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

Returns the actor's return value, validated through :attr:result_adapter. The result type is R exactly — never R | None. Missing or failed results raise.

Raises:

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

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

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

    Returns the actor's return value, validated through
    :attr:`result_adapter`. The result type is ``R`` exactly —
    never ``R | None``. Missing or failed results raise.

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

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

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

        await asyncio.sleep(sleep)

progress_stream async

progress_stream() -> AsyncIterator[ProgressEvent]

Stream live progress events for this job.

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

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

Yields events until a terminal=True event is produced.

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

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

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

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

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

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

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