Actor API¶

An actor is an async handler registered with the worker. The @actor decorator introspects the handler signature, validates it at decoration time, and returns an ActorRef[P, R] that carries the actor's payload and result types end-to-end.

Contents¶

The @actor decorator
Handler signatures
Sync actors
ActorRef[P, R]
JobContext[P]
Payload and result types
Singleton actors
unique_for deduplication
max_pending backpressure
Rate limits and reservations
Retry policy
Control-flow exceptions
Dependency injection
Sub-job enqueuing
Progress reporting
Testing actors without a database
Full worked example
Actor deregistration

The `@actor` decorator¶

Supports both plain and parameterised forms:

from taskq import RetryPolicy, actor


# Plain: all options take their defaults.
@actor
async def send_email(payload: EmailPayload) -> EmailResult: ...


# Parameterised.
@actor(queue="priority", max_concurrent=10, retry=RetryPolicy(max_attempts=5))
async def process_order(payload: OrderPayload) -> OrderResult: ...

Parameters¶

Parameter	Type	Default	Description
`name`	`str \\| None`	`fn.__qualname__`	Actor name stored in the `actor_config` table and on every job row. Override when the qualified name would be unstable across refactors.
`queue`	`str`	`"default"`	Queue this actor is dispatched on. Must match `[A-Za-z0-9_][A-Za-z0-9_.-]*` (note `:` is excluded; it separates segments in the `taskq:global:queue:` concurrency-cap namespace. Can be overridden per-enqueue.
`retry`	`RetryPolicy \\| None`	`RetryPolicy()`	Retry policy; see Retry policy. `None` resolves to the default `RetryPolicy()`.
`result_ttl`	`timedelta \\| None`	`None`	How long the result JSONB is retained after a job succeeds. `None` means retain indefinitely. Only the seed value: once a worker has synced this actor once, the stored `actor_config.result_ttl` is authoritative and this literal is ignored; see ActorConfig sync.
`singleton`	`bool`	`False`	Enforce at most one active job of this actor fleet-wide; see Singleton actors.
`max_concurrent`	`int \\| None`	`None`	Fleet-wide concurrency cap, best-effort, not a hard cap. `None` = unbounded. `0` = drain mode (no jobs dispatched). May exceed the configured value by up to `(num_active_producers - 1) * max_concurrent`; use a reservation for strict enforcement; see Choosing a concurrency control. Only the seed value: once a worker has synced this actor once, the stored `actor_config.max_concurrent` is authoritative and this literal is ignored; tune it live with `taskq actor-config set`, see ActorConfig sync.
`max_pending`	`int \\| None`	`None`	Queue-depth backpressure cap; see `max_pending` backpressure. Only the seed value: once a worker has synced this actor once, a non-NULL stored `actor_config.max_pending` is authoritative and this literal is ignored; tune it live with `taskq actor-config set`, see ActorConfig sync.
`metadata`	`dict[str, object] \\| None`	`{}`	Arbitrary key-value metadata stored in `actor_config.metadata` (JSONB). Must be a plain `dict`; mapping proxies and frozendicts are rejected at decoration time. The key `"singleton"` is reserved by the library.
`unique_for`	`timedelta \\| None`	`None`	Deduplication window; see `unique_for` deduplication.
`unique_states`	`tuple[JobStatus, ...]`	`("pending", "scheduled", "running", "succeeded")`	Job statuses a `unique_for` window matches. `succeeded` is included because it is the state that says the work already happened; the failure states are excluded so one failure does not suppress the identity for the rest of the window.
`start_to_close`	`timedelta \\| None`	`None`	Per-attempt execution timeout. Precedence (first wins): per-enqueue `start_to_close` > this actor default > `TASKQ_DEFAULT_START_TO_CLOSE`. `None` means no per-attempt timeout unless a worker-wide default is set. See Retries: `start_to_close` vs `schedule_to_close`.
`rate_limits`	`list[str \\| KeyedRateLimitRef \\| TokenBucket \\| SlidingWindow] \\| None`	`[]`	Named rate-limit buckets this actor consumes; see Rate limits and reservations.
`reservations`	`list[str \\| KeyedReservationRef \\| ConcurrencyReservation] \\| None`	`[]`	Named concurrency reservation slots this actor claims. A `KeyedReservationRef` derives per-key (session/tenant) reservation buckets from the job payload at dispatch time; see Rate limits and reservations.
`non_retryable_exceptions`	`tuple[type[BaseException], ...]`	`()`	Exception types that fail the job immediately instead of retrying.
`retry_classifier`	`RetryClassifierHook \\| None`	`None`	Hook for exception-instance-level retry classification. Invoked with `(exception, attempt)` for exceptions that survive `non_retryable_exceptions`/`PayloadValidationError` checks; return `RetryOverride` to refine `kind`/`delay` per occurrence or `None` to fall back to the static `RetryPolicy`. See Retries: `retry_classifier` hook.
`on_retry_exhausted`	`OnRetryExhausted \\| None`	`None`	Callback invoked when the retry budget is exhausted, after the job is marked `failed`.
`on_retry_exhausted_timeout`	`float`	`3.0`	Seconds allowed for `on_retry_exhausted` to complete before it is abandoned.
`on_success`	`OnSuccess \\| None`	`None`	Callback invoked when the job succeeds, after the transaction commits. Receives `(job_row, result)`. Mirrors `on_retry_exhausted` with a timeout guard; see Retries: `on_success` hook.
`on_success_timeout`	`float`	`3.0`	Seconds allowed for `on_success` to complete before it is abandoned.
`on_cancel`	`OnCancel \\| None`	`None`	Callback invoked when a running job ends `cancelled`, beside the terminal write. Receives `(job_row)`: a cancelled attempt produced no result. Best-effort and timeout-bounded like the hooks above; see Job Cancellation: `on_cancel` hook.
`on_cancel_timeout`	`float`	`3.0`	Seconds allowed for `on_cancel` to complete before it is abandoned.
`priority`	`int`	`0`	Default dispatch priority for jobs enqueued without an explicit `priority=`. Must fit `smallint` range (-32768..32767).

on_cancel cannot fire for a job cancelled before it ran

The hook runs on the worker that was executing the attempt. A job cancelled while still pending or scheduled never reached a worker, so no hook of any kind runs for it and none can: there is no attempt to clean up after. Releasing whatever such a job had reserved before it was enqueued stays with whoever issued the cancel. This matters because the cancel an operator issues most often, on a job sitting in the queue, is exactly the one the hook cannot see, so cleanup that must happen for every cancellation belongs on the caller's side of the enqueue, not in on_cancel.

Tell the two apart on the row: a cancel that reached a running actor stamps error_class = 'CancelledCooperatively' (or 'CancelledForced' once the escalation ladder had to interrupt it), one that never started stamps 'CancelledBeforeStart', and an actor that had to be taken away after the cancellation graces stamps 'CancelAbandoned'.

max_concurrent, max_pending and result_ttl are seed-only: changing the literal does nothing on an existing deployment

These three are operator-owned once a row exists. The startup UPSERT omits them from its DO UPDATE SET clause, so Postgres leaves the stored value untouched and the decorator literal is read only when the actor_config row is first created. On any deployment where the actor has booted even once, editing @actor(max_concurrent=...) and shipping it is a no-op: the code says one thing and dispatch does another, with nothing failing to draw your attention to it.

The tell is an info-level actor-config-capacity-override log line at startup, one per actor whose literal differs from the stored row.

Check what is actually stored, then set it:

taskq actor-config diff          # registered literal vs stored row, per actor
taskq actor-config get send_email
taskq actor-config set send_email --max-concurrent 4   # takes effect next dispatch cycle

A first deploy seeds whatever the literal says, so this trap only bites on the second change, and it bites silently. If you want decorator literals to be authoritative, reconcile them explicitly on boot with taskq actor-config set (it is idempotent) rather than assuming the decorator did it. See ActorConfig sync.

A NULL stored max_concurrent means uncapped, not "fall back to the literal", so an actor seeded before you added the argument stays uncapped until an operator sets it.

Choosing a concurrency control¶

TaskQ limits concurrency at several independent scopes. They are not interchangeable, and the strict ones are the leased-slot mechanisms, not the max_concurrent decorator argument.

You want to…	Use	Scope	Strict?
Protect one actor that must not run in parallel (thread-unsafe library, single GPU, non-reentrant native code)	`@actor(reservations=[...])` with a `ConcurrencyReservation`	Fleet-wide, per named slot bucket	Yes (leased slot rows)
Bound a whole queue fleet-wide regardless of which actors publish to it	`taskq queues set-max-concurrent <queue> --max-concurrent N`	Fleet-wide, per queue	Yes (same leased-slot machinery)
Damp one actor's share of the fleet without needing an exact bound	`@actor(max_concurrent=N)` / `taskq actor-config set`	Fleet-wide, per actor	No (best-effort)
Size one process (how many jobs a single worker runs at once)	`TASKQ_MAX_CONCURRENCY`	Per worker process	Per-process only

Only a reservation is strict. max_concurrent, singleton and identity_key are not.

actor_config.max_concurrent is a per-round admission damper. Dispatch reads the running count once, before it takes FOR UPDATE SKIP LOCKED row locks, and never rechecks it. Concurrent dispatchers each see the same count, each admit up to the residual, and lock disjoint rows, so SKIP LOCKED does not serialize them and every one of them succeeds. The over-dispatch bound is (num_producers - 1) * max_concurrent per round, and those jobs genuinely run: reclaiming stale locks does not undo an over-dispatch.

identity_key does not fix this. The dispatch CTE's running_identities snapshot is read in the same statement, before the same row locks, so two dispatchers can each see an identity as "not running" and both admit a job for it. It is bounded by roughly the number of concurrent dispatchers per identity per round, not a hard 1. max_concurrent=1 plus an identity_key gives you approximately-one-per-identity, which is useful for fairness and useless as a mutex.

singleton=True is a different guarantee, not a stronger one. It is enforced at enqueue time by a partial unique index, is actor-scoped rather than identity-scoped, and rejects the producer with SingletonCollisionError instead of queueing. It bounds how many jobs exist, not how many run in parallel on a given resource.

If two concurrent executions would corrupt data, none of the above is sufficient: use a reservation.

Worked example: "I need a real mutex"¶

A thread-unsafe PDF library where two concurrent calls corrupt output. One slot means one job at a time, fleet-wide, enforced by a physical row:

from datetime import timedelta

from taskq import actor
from taskq.ratelimit import ConcurrencyReservation


# slots=1 → a genuine fleet-wide mutex. Acquire is a single
# read-and-write statement against one slot row, so there is no
# read-then-decide window for a second dispatcher to slip through.
@actor(
    reservations=[
        ConcurrencyReservation(
            name="pdf_render",
            slots=1,
            lease=timedelta(minutes=5),  # must exceed worst-case runtime
        )
    ],
)
async def render_pdf(payload: RenderPayload) -> RenderResult:
    return await do_render(payload)  # guaranteed sole occupancy

Declaring the instance inline needs no separate registration step; the worker registers it and pre-allocates its slot rows at bootstrap. See Wiring to Actors; use sync_slots when changing slots on a running deployment.

Two things to get right:

lease must exceed the job's worst-case runtime. The heartbeat loop renews the lease while the job runs, but a job that outlives its lease without heartbeating has its slot swept as leaked and a second job can acquire it.
ReservationUnavailable is raised when every slot is held. Jobs are retried rather than run without a slot, so the cap never degrades into best-effort under load.

Do not size a queue for its narrowest actor

A queue cap bounds a tier, and cannot tell a PDFium call from an HTTP round-trip. Throttling a queue to 1 to protect one thread-unsafe actor also serializes every unrelated actor sharing that queue: a fleet that looks configured and produces almost nothing. Put the limit on the actor that needs it (a reservation), or move that actor to its own queue.

Decoration-time validation¶

The decorator raises at import time (not at runtime) when:

The payload parameter lacks a type annotation.
The return annotation is missing.
The payload annotation is not a pydantic.BaseModel subclass.
A JobContext parameter's payload type does not match the handler's payload type.
An unannotated DI parameter is declared.
max_concurrent is a negative integer.
max_pending is a negative integer.
metadata is not a plain dict or None.
singleton is not a bool.

Both async def and plain def functions are accepted. Sync functions are dispatched via asyncio.to_thread() and the actor's is_sync property is True.

Handler signatures¶

The decorator accepts any of these four shapes. Declare only what the handler body needs.

Payload only¶

from pydantic import BaseModel
from taskq import actor


class ResizePayload(BaseModel):
    image_id: str
    width: int
    height: int


class ResizeResult(BaseModel):
    url: str


@actor
async def resize_image(payload: ResizePayload) -> ResizeResult:
    # No ctx, no DI deps injected.
    return ResizeResult(url=f"https://cdn.example.com/{payload.image_id}")

Payload and context¶

from taskq import Snooze, actor
from taskq.context import JobContext


@actor
async def resize_image(payload: ResizePayload, ctx: JobContext[ResizePayload]) -> ResizeResult:
    if ctx.cancellation_requested:
        raise Snooze(timedelta(minutes=1))
    return ResizeResult(url=f"https://cdn.example.com/{payload.image_id}")

Payload and DI dependencies¶

from taskq import actor


@actor
async def resize_image(
    payload: ResizePayload,
    *,
    db: DbSession,
    http: HttpClient,
) -> ResizeResult:
    row = await db.fetch_one("SELECT * FROM images WHERE id = $1", payload.image_id)
    ...

DI parameters must be keyword-only (* separator) or positional; the worker always passes them as keyword arguments. Every DI parameter must have a type annotation that is a concrete class; the worker's DI resolver maps the annotation to a registered provider at dispatch time.

Payload, context, and DI dependencies¶

from taskq import actor
from taskq.context import JobContext


@actor(queue="priority")
async def resize_image(
    payload: ResizePayload,
    ctx: JobContext[ResizePayload],
    *,
    db: DbSession,
) -> ResizeResult:
    ctx.log.info("starting", attempt=ctx.attempt)
    ...

Sync actors¶

@actor also accepts plain def functions. Sync actors run in a thread via asyncio.to_thread(), so the event loop is never blocked.

from pydantic import BaseModel
from taskq import actor


class PdfPayload(BaseModel):
    html: str
    filename: str


class PdfResult(BaseModel):
    s3_key: str


@actor(queue="media")
def generate_pdf(payload: PdfPayload) -> PdfResult:
    # CPU-bound PDF generation: runs in a thread, not the event loop.
    import weasyprint

    out = weasyprint.HTML(string=payload.html).write_pdf()
    # ... upload to S3 ...
    return PdfResult(s3_key=f"pdfs/{payload.filename}")

Cancellation for sync actors¶

Sync actors cannot be force-cancelled via asyncio.Task.cancel(). They must cooperate by polling ctx.should_abort():

@actor
def long_loop(payload: BigPayload, ctx: JobContext[BigPayload]) -> None:
    for item in payload.items:
        if ctx.should_abort():
            raise asyncio.CancelledError  # cooperative exit; job marked cancelled
        process(item)

Phase 1 (COOPERATIVE): ctx.should_abort() returns True. The actor must RAISE to cancel: returning normally after a cancel request marks the job succeeded (the attempt's outcome is what the actor's return says it is; the consumer never second-guesses a returned value), so cancelled work would silently report success. Raising asyncio.CancelledError (or calling ctx.check_cancelled(), which raises it for you) is what routes the attempt to cancelled.
Phase 2 (FORCED): The cancel controller writes cancel_phase=2 to PG but cannot interrupt the thread. The sync actor continues until it polls should_abort() or hits start_to_close timeout.
Phase 3 (ABANDON): If the actor never polls, the job is abandoned after cancel_grace + cleanup_grace.

Sync actors and shutdown¶

SIGTERM (or SIGINT) triggers the worker's graceful shutdown (workers.md — Graceful shutdown). For a sync actor the contract has one hard fact at its center: the thread is uncancellable. A sync actor runs on the default executor via asyncio.to_thread() (src/taskq/actor.py), and task.cancel() — what the shutdown's FORCING phase does — cancels the await, never the thread; CPython offers no API to interrupt a running thread. What each phase actually does to a sync actor:

CANCELLING sets the cancel event, so ctx.should_abort() starts returning True.
FORCING cancels the job's asyncio task: the await unwinds, the actor's body keeps running to its own end.
RELEASING does not assume the actor unwound with the await. The thread's handle is exit-tracked (_run_sync_actor_tracked, src/taskq/worker/dispatch.py), and the row's release is held back until the thread has provably exited or the exit budget is spent: held scheduled behind the remaining termination_grace_period plus the watchdog's exit tail, capped by lock_lease — the fleet never gets a row whose actor is still running in this process.
The exit is enforced, not hoped for. The shutdown watchdog stays armed until every tracked actor handle is reaped: either the thread finishes on its own, or the deadline trip fires and os._exit ends the process with the still-running thread inside it (the trip's reason is tracked-actor-outlived-teardown). A third signal while that gate is waiting also exits by join: a running actor thread is waited out, not killed.

What the job body must do:

Poll ctx.should_abort() in long loops and exit promptly when it turns True. RAISE to cancel (asyncio.CancelledError, or ctx.check_cancelled()), per the raise-vs-return rule in the cancellation subsection above: an actor that exits inside the grace gets the immediate pending release.
Budget the grace. Put a should_abort() poll at least every few seconds, so the body lands well inside cancellation_grace_period (30s at the defaults). An actor that cannot check in still gets the held release — the row is never stranded — but every deploy it outlives pays the exit window's latency, and at the deadline trip the thread dies with the process, by design.
Prefer async actors for cancellable work. An async def actor unwinds with the await: task.cancel() reaches the body's current await directly, and the row is released pending immediately. Reach for a sync actor for what threads are actually for — blocking libraries that release the GIL — not as the default shape.

What the body must not do:

Block indefinitely on a call that never returns: no poll opportunity means the grace cannot reach you.
Install its own signal handlers or fight the worker's shutdown ordering.
Fan out threads or subprocesses and expect shutdown to reap them: "provably exited" is wrapper-level, about TaskQ's own executor thread. A subprocess the actor spawned outlives even os._exit.

The heartbeat-loss residual. On the isolate path (isolate_self, src/taskq/worker/heartbeat.py) — the walk-away a worker takes when Postgres stops answering its heartbeat — locally-running actors get the same SHUTDOWN cancel stamp and task cancel, and the path joins them bounded (cancellation_grace_period + cleanup_grace_period plus close slack). A wedged sync thread outlives that join: its row stays excluded from the re-pend, and when its consumer's unwinding never completes the interrupt write never lands, so the row's release falls to the lock-lease expiry backstop (workers.md — Heartbeat and liveness).

DI thread safety¶

DI resolution happens in the event loop before the sync function is dispatched to the thread. Resolved kwargs are passed through:

Scope	Safe in sync actor?	Notes
PROCESS	Depends on object	Shared across event loop and thread
LOOP	NOT safe	`asyncpg.Connection`, `redis.asyncio.Redis` are not thread-safe
TRANSIENT	Safe if object is thread-safe	Fresh per invocation

The worker logs a WARNING when a sync actor declares a LOOP-scoped dependency parameter. For thread-safe database access, register a sync driver connection at Scope.THREAD or use Scope.TRANSIENT.

Rate limiting and sub-jobs¶

Rate-limit acquisition/release runs in the event loop before/after asyncio.to_thread(). No changes needed.

Sub-job enqueues from a sync actor use the autonomous commit path (acquires a fresh pool connection). The transactional LOOP-scope connection path is unavailable from threads.

`ActorRef[P, R]`¶

ActorRef[P, R] is the object returned by @actor. It is not a callable that enqueues jobs; pass it to JobsClient.enqueue for that. Direct in-process invocation (await my_actor(payload, ...)) runs the handler without going through the queue and is intended for tests and simulators.

Properties¶

Property	Type	Description
`name`	`str`	Actor name (from `name=` or `fn.__qualname__`).
`queue`	`str`	Default queue.
`payload_type`	`type[P]`	The Pydantic model class for `P`. Used to validate raw payloads at dispatch time.
`result_adapter`	`TypeAdapter[R]`	Pydantic `TypeAdapter` for `R`. Serialises the result to JSONB on the worker side and deserialises it in `JobHandle.wait()`.
`retry`	`RetryPolicy`	The actor's retry policy.
`result_ttl`	`timedelta \\| None`	Result retention window.
`singleton`	`bool`	Whether singleton enforcement is active.
`max_concurrent`	`int \\| None`	Fleet-wide concurrency cap.
`max_pending`	`int \\| None`	Queue-depth backpressure cap.
`metadata`	`dict[str, object]`	Actor-level metadata.
`unique_for`	`timedelta \\| None`	Deduplication window.
`unique_states`	`tuple[JobStatus, ...]`	Active statuses for dedup.
`rate_limits`	`list[str \\| KeyedRateLimitRef \\| TokenBucket \\| SlidingWindow] \\| None`	Rate-limit bucket names, keyed refs, or bucket objects.
`reservations`	`list[str \\| KeyedReservationRef \\| ConcurrencyReservation] \\| None`	Concurrency reservation names, keyed refs, or reservation objects.
`wants_ctx`	`bool`	Whether the handler declared a `JobContext` parameter.
`is_sync`	`bool`	`True` when the handler is a plain `def` (not `async def`). Sync actors run via `asyncio.to_thread()`.
`dependencies`	`dict[str, type[object]]`	DI parameter names mapped to their annotated types.
`fn`	`Callable[..., object]`	The underlying handler (sync or async). Prefer `__call__` for invocation.

Direct invocation (`call`)¶

# Handler with no ctx:
result = await resize_image(ResizePayload(image_id="abc", width=800, height=600))

# Handler with ctx (supply a real or stub JobContext):
result = await resize_image(payload, ctx, db=db_stub)

Passing ctx to a no-ctx handler raises TypeError. Omitting ctx for a ctx handler also raises TypeError. Missing DI dependencies surface as TypeError from Python's argument binding.

`JobContext[P]`¶

JobContext[P] is a frozen dataclass constructed per attempt by the worker. Handlers receive it as the second positional parameter when they declare ctx: JobContext[YourPayload].

Fields¶

Field	Type	Description
`job_id`	`UUID`	The job's unique identifier.
`actor`	`str`	Actor name.
`queue`	`str`	Queue the job is running on.
`attempt`	`int`	Current attempt number (1-indexed).
`worker_id`	`UUID`	The worker running this attempt.
`payload`	`P`	Fully-validated payload instance. Typed as `P`; no cast required.
`jobs`	`SubJobEnqueuer`	Enqueue sub-jobs from within the actor body; see Sub-job enqueuing.
`log`	`structlog.BoundLogger`	Structured logger pre-bound with `job_id`, `actor`, and `attempt`.
`span`	`opentelemetry.trace.Span \\| None`	Active OTel span for this attempt. `None` when tracing is disabled.
`cancel_event`	`asyncio.Event`	Set by the cancel-poll hook when the job enters cooperative cancellation.
`progress(...)`	`async method`	Report incremental progress for this job; see Progress reporting.
`cancellation_requested`	`bool` (property)	Returns `True` when `cancel_event` is set.
`check_cancelled()`	`method → None`	Raises `asyncio.CancelledError` if `cancel_event` is set. Convenience for cooperative exit inside actor loops.
`should_abort()`	`method → bool`	Sync-only. Thread-safe cooperative cancellation check. Returns `True` when cancellation has been requested. Sync actors must poll this; they cannot `await cancel_event.wait()` from a thread.

`cancellation_requested` property¶

@property
def cancellation_requested(self) -> bool: ...

Returns True when cancel_event is set, meaning a cancellation request has reached phase 1 (cooperative). Poll this in long-running loops to exit cleanly:

@actor
async def long_job(payload: LongPayload, ctx: JobContext[LongPayload]) -> None:
    for item in payload.items:
        if ctx.cancellation_requested:
            raise Snooze(timedelta(minutes=5))
        await process(item)

Alternatively, await ctx.cancel_event.wait() blocks until cancellation is requested.

Payload and result types¶

Both P and R must be pydantic.BaseModel subclasses. R may additionally be None for fire-and-forget actors.

from pydantic import BaseModel


class OrderPayload(BaseModel):
    order_id: str
    items: list[str]
    total_cents: int


class OrderResult(BaseModel):
    confirmation_number: str
    estimated_delivery: str


@actor
async def process_order(payload: OrderPayload) -> OrderResult: ...


# Fire-and-forget: R = None
@actor
async def audit_log(payload: AuditPayload) -> None: ...

Plain dict, dataclass, and TypedDict are not supported as payload or result types. Pydantic v2 models are required for JSONB round-trip serialisation.

Singleton actors¶

singleton=True enforces at most one active job of this actor fleet-wide across all queues and workers.

@actor(singleton=True)
async def daily_report(payload: ReportPayload) -> None: ...

Semantics:

"Active" means status IN ('pending', 'scheduled', 'running'). A snoozed singleton job in scheduled state blocks new enqueues until it terminates.
Singleton enforcement is actor-scoped, not identity-scoped. Different identity_key values for the same singleton actor are still blocked.
For per-identity singleton semantics (one active job per user, not per actor), use max_concurrent=1 with an identity_key instead, but note this is best-effort, not a hard 1: the dispatcher's running_identities snapshot is read before it takes row locks, so concurrent dispatchers can each admit a job for the same identity. Use it for fairness, never as a mutex. For a strict guarantee use a ConcurrencyReservation; a KeyedReservationRef gives one strict slot bucket per key.
The library injects metadata["singleton"] = True on every enqueue. Callers must not set this key manually; the library unconditionally overwrites it.
On collision, SingletonCollisionError is raised.

from taskq.exceptions import SingletonCollisionError

try:
    handle = await client.enqueue(daily_report, ReportPayload(date="2025-01-01"))
except SingletonCollisionError as exc:
    print(f"blocked by job {exc.blocking_job_id}, retry_after={exc.retry_after}")

SingletonCollisionError.blocking_job_id is the UUID of the existing job from the pre-flight query, or None on the race path (Layer 2 unique constraint catch). retry_after is derived from the blocking job's schedule_to_close when available, otherwise None.

`unique_for` deduplication¶

unique_for deduplicates enqueues for the same (actor, identity_key) within a sliding window. For how to choose between this, idempotency_key, and singleton: they are three different tools; see the chooser table in ops.md: Fan-out at scale.

from datetime import timedelta


@actor(
    unique_for=timedelta(minutes=15),
    unique_states=("pending", "scheduled", "running"),
)
async def sync_account(payload: SyncPayload) -> None: ...

Semantics:

unique_for only has effect when identity_key is also provided at enqueue time. If identity_key is omitted, unique_for is a silent no-op; the library logs a warn-once actor_config_unique_for_ignored (on every enqueue surface: an actor-declared unique_for through JobsClient/TaskQ, and a per-call unique_for through SubJobEnqueuer) and creates a fresh job every time. This is a common footgun: configure unique_for on the actor but forget to pass identity_key at the call site.
Deduplication is single-flight: a transaction-scoped advisory lock serializes the preflight-then-insert per (actor, identity_key) on pool and bare-caller connections alike, so concurrent enqueues dedupe against the winner's row rather than both inserting; the dispatch CTE's running_identities filter remains as the backstop for cross-window races.
When a dedup match is found, JobHandle.was_existing is True and the handle wraps the existing job row.
unique_states controls which statuses the window check matches. The default is ("pending", "scheduled", "running", "succeeded"): the window means "at most one job for this identity in this period", and succeeded is the state that says the work already happened: without it the identity would be free again the instant the first job completed, so a re-delivered webhook or a double-clicked button inside a still-open window would run the work a second time. The failure states (failed, cancelled, crashed, abandoned) are excluded because they mean the work did not happen; matching them would let one transient failure suppress every later attempt for the rest of the window. To block only concurrent execution, pass unique_states=("pending", "scheduled", "running") explicitly.
A dedup onto a job that is already finished is reported: the enqueue logs a WARN-level enqueue_deduplicated line naming the matched status, and JobHandle.deduplicated_onto_terminal is True so the caller can branch without re-reading the row.

handle = await client.enqueue(
    sync_account,
    SyncPayload(account_id="acct_123"),
    identity_key="acct_123",  # required; unique_for is a no-op without this
)
if handle.was_existing:
    print("deduped: returning existing job handle")

The identity_key can be passed per-enqueue via JobsClient.enqueue. (A per-call unique_for override exists only on SubJobEnqueuer.enqueue, inside an actor body; JobsClient.enqueue always uses the actor-declared window.)

`max_pending` backpressure¶

max_pending limits the number of pending + scheduled jobs before enqueue rejects.

@actor(max_pending=1000)
async def ingest_event(payload: EventPayload) -> None: ...

Semantics:

None (default) means unbounded: enqueue never rejects on capacity.
max_pending=0 means no jobs are ever accepted (every enqueue raises immediately).
Negative values raise ValueError at decoration time.
When the limit is reached, MaxPendingExceededError is raised synchronously. The caller decides whether to retry, back off, or drop.

from taskq.exceptions import MaxPendingExceededError

try:
    handle = await client.enqueue(ingest_event, EventPayload(data=raw))
except MaxPendingExceededError as exc:
    print(f"queue full: {exc.current_count}/{exc.max_pending} pending for {exc.actor}")

Evaluation order at enqueue: unique_for dedup → singleton pre-flight → max_pending count check → idempotency_key upsert → job INSERT. A unique_for hit bypasses all remaining checks. A singleton collision fires before max_pending.

Operator-owned limit. The enforced limit is the effective max_pending: a non-NULL stored actor_config.max_pending (set with taskq actor-config set <actor> --max-pending N) wins over this literal; a cleared (--clear-max-pending) or absent stored value falls back to the literal. Enqueue-side processes read the stored value through a TTL-bounded cache (default 5s), so an operator change takes effect fleet-wide within seconds, with no redeploy, no worker restart. Run taskq actor-config diff --actors myapp.actors:registry to see the literal, the stored value, and which one is currently enforced. See ActorConfig sync.

Per-call argument. SubJobEnqueuer.enqueue(..., max_pending=N) (inside an actor body) is resolved against the effective limit, not in place of it: against a stored cap the tighter of the two wins (min(stored, N)), an explicit caller shedding load is never widened by an operator override, and no code path can raise the operator's fleet cap. With nothing stored, the per-call argument wins outright over this literal, in both directions: actor code may loosen its own declaration.

Rate limits and reservations¶

Declare named rate-limit buckets and concurrency reservation slots on the actor:

@actor(
    rate_limits=["openai", "vendor_x"],
    reservations=["gpu_pool"],
)
async def run_inference(payload: InferencePayload) -> InferenceResult: ...

rate_limits and reservations are lists of bucket/slot names defined in the rate-limiting configuration. For bucket and slot configuration syntax see Rate Limiting.

Retry policy¶

Pass a RetryPolicy to control how the worker retries failed jobs.

from datetime import timedelta
from taskq import actor
from taskq.retry import RetryPolicy


@actor(
    retry=RetryPolicy(
        kind="transient",  # "transient" | "indefinite" | "non_retryable"
        max_attempts=5,  # ignored for kind="indefinite"
        backoff="exponential",  # "exponential" | "linear" | "fixed"
        base=timedelta(seconds=10),
        cap=timedelta(hours=2),
        jitter=0.2,
        time_budget=None,  # only used for kind="indefinite"
    )
)
async def flaky_call(payload: CallPayload) -> CallResult: ...

`RetryPolicy` fields¶

Field	Type	Default	Description
`kind`	`"transient" \\| "indefinite" \\| "non_retryable"`	`"transient"`	Retry tier. `"transient"` retries up to `max_attempts`. `"indefinite"` retries until `time_budget` or `schedule_to_close` elapses. `"non_retryable"` never retries.
`max_attempts`	`int`	`3`	Maximum attempts. Must be >= 1. Used only when `kind="transient"`.
`time_budget`	`timedelta \\| None`	`None`	Total wall-clock budget. Used only when `kind="indefinite"`.
`backoff`	`"exponential" \\| "linear" \\| "fixed"`	`"exponential"`	Backoff shape.
`base`	`timedelta`	`5s`	Base delay for backoff computation. Must be > 0.
`cap`	`timedelta`	`1h`	Maximum per-attempt delay (must be >= `base`).
`jitter`	`float`	`0.2`	Multiplicative jitter factor in `[0.0, 1.0]`. Applies symmetric jitter: `delay * uniform(1-jitter, 1+jitter)`.

For retry internals (decision logic, backoff formula, on_retry_exhausted hook) see Worker.

Control-flow exceptions¶

These exceptions are signals, not errors. Raise them from an actor body to drive state transitions without consuming retry budget (for Snooze) or while consuming it (for RetryAfter).

`Snooze`¶

Reschedules the job at now + delay without consuming retry budget. The job re-enters scheduled state.

from datetime import timedelta
from taskq.exceptions import Snooze


@actor
async def poll_external_api(payload: PollPayload, ctx: JobContext[PollPayload]) -> None:
    result = await check_status(payload.task_id)
    if result.status == "pending":
        raise Snooze(timedelta(minutes=2))
    # ... handle completion

Snooze(delay) raises ValueError if delay < timedelta(0).

A snoozed singleton actor re-enters scheduled state and blocks new enqueues until it either succeeds, fails, or is cancelled.

`RetryAfter`¶

Schedules a retry at a specific delay. Consumes the retry budget by default.

from datetime import timedelta
from taskq.exceptions import RetryAfter


@actor
async def call_rate_limited_api(payload: ApiPayload) -> ApiResult:
    response = await api_client.call(payload.endpoint)
    if response.status == 429:
        retry_in = timedelta(seconds=int(response.headers.get("Retry-After", 60)))
        raise RetryAfter(retry_in)
    return ApiResult(data=response.json())

RetryAfter(delay, consume_budget=False) reschedules without counting the attempt against max_attempts. consume_budget=True is the default.

RetryAfter(delay) raises ValueError if delay < timedelta(0).

Dependency injection¶

Actors declare DI dependencies as keyword-only parameters. The worker resolves them from a ProviderRegistry at dispatch time.

Registering providers¶

Create a ProviderRegistry, register your providers, validate it, then pass it to the worker. See Dependency Injection for the full wiring. The three registration methods are:

from taskq.di import ProviderRegistry, Scope

registry = ProviderRegistry()

# Register a pre-built singleton value (PROCESS scope: lives for the
# duration of the worker process).
registry.register_value(Database, Scope.PROCESS, db_instance)


# Register an async factory. The factory is called once per scope lifetime.
# Use an async generator to run teardown code.
async def create_http_client():
    client = HttpClient()
    try:
        yield client
    finally:
        await client.aclose()


registry.register_factory(HttpClient, Scope.LOOP, create_http_client)

# Register a class; the DI engine instantiates it and detects lifecycle
# methods (aclose, close) automatically.
registry.register_class(MyService, Scope.LOOP)

# Validate and seal before starting the worker.
registry.validate(actors=[my_actor])

After validate() the registry is sealed; further registrations raise RuntimeError.

Scope lifetimes¶

Scope	Lifetime	Typical use
`Scope.PROCESS`	Worker process start to exit	Config, shared read-only singletons
`Scope.LOOP`	Event loop start to loop close	asyncpg pools, HTTP clients, Redis clients
`Scope.TRANSIENT`	Per actor invocation	Per-request helpers, one-shot contexts

A provider may depend only on providers of the same or wider scope. Violations raise ScopeViolation at validate() time.

Declaring dependencies in an actor¶

@actor
async def send_email(
    payload: SendEmailPayload,
    *,
    db: Database,
    http: HttpClient,
) -> None:
    record = await db.get(payload.recipient_id)
    await http.post("/send", json={"to": record.email})

payload and ctx are always supplied by the consumer and must not be registered as providers. All other annotated keyword parameters are resolved from the registry. Missing providers raise MissingProvider at validate() time, not at runtime.

Sub-job enqueuing¶

Enqueue sub-jobs from within an actor body via ctx.jobs, which is a SubJobEnqueuer.

@actor
async def process_batch(payload: BatchPayload, ctx: JobContext[BatchPayload]) -> None:
    for item_id in payload.item_ids:
        await ctx.jobs.enqueue(
            process_item,
            ItemPayload(item_id=item_id),
            priority=1,
        )

See the SubJobEnqueuer reference in Client API: SubJobEnqueuer.

Transaction semantics¶

Sub-job enqueues join the parent job's transaction by default. On a TASKQ_MAX_CONCURRENCY=1 worker that transaction runs on the registered LOOP-scope asyncpg.Connection; at higher concurrency the worker opens a dedicated per-slot transaction pool, each job transacts on its own slot connection, and the actor receives that same slot connection by injection, so the actor's own database writes join the job's transaction too, and two concurrent slots' actors can never interleave operations on one connection (asyncpg permits one operation per connection at a time). Either way, sub-job INSERTs are part of the parent's database transaction:

If the parent actor succeeds, the transaction commits and the sub-jobs become visible.
If the parent actor raises an exception (and will be retried or failed), the transaction rolls back and the sub-jobs vanish atomically; they are never seen by the queue.

This is the correct default for fan-out patterns where sub-jobs should only exist if the parent completes successfully.

The post-commit flush can still fail. On the transactional path the actor's sub-enqueues are buffered and replayed against the backend only after the parent's transaction has committed and the parent has been marked succeeded. If one or more of those replayed enqueues fails (a batch row that went terminal mid-flight, a capacity refusal, a driver error), flush_buffer() collects every failure and raises SubEnqueueError — the parent job is already succeeded, so this is the signal that child jobs were lost: the caller believes work exists that was never durably enqueued. SubEnqueueError.failed_items carries each failed EnqueueArgs together with the exception that sank it; re-enqueue from it. See the Exceptions API reference.

Tag inheritance. Sub-jobs enqueued via ctx.jobs.enqueue() inherit the parent job's tags by default. This makes sub-jobs findable by JobFilter(tags=...) and cancellable by cancel_where. Pass inherit_tags=False to suppress inheritance for a specific sub-job, or pass explicit tags=[...] to merge with inherited tags. See Jobs & Clients: Sub-job tag inheritance for the full semantics table.

Autonomous fallback. If no LOOP-scope asyncpg.Connection is registered in the DI container, ctx.jobs.enqueue() falls back to the worker pool and commits each INSERT independently. In this mode, sub-jobs are persisted even if the parent subsequently raises an exception. The worker emits a sub_enqueue_autonomous_fallback warning to structlog every 100 autonomous enqueues to alert you that transactional guarantees are not in effect.

To ensure the transactional path is active, register an asyncpg.Connection at Scope.LOOP in the DI registry (see Dependency Injection). Registering an asyncpg.Pool at Scope.LOOP instead does not activate it: the transactional path keys on Connection, so a Pool-only registration keeps the autonomous fallback in force, transactional consume is silently disabled, not enabled.

Transactional consume: session state, not concurrency, is the constraint

At max_concurrency > 1 each job (actor writes, terminal write, and transactional sub-enqueues alike) transacts on its own per-slot connection, so the transactional path is correct at any concurrency. The registered LOOP-scope connection is that mode's activation signal; the slot connections are fresh direct-DSN connections that never carried its session state. If your writes depend on session state (SET ROLE, search_path, an RLS-driving GUC) set up on the registered connection, run the transactional actor on a TASKQ_MAX_CONCURRENCY=1 worker, where the transaction the actor's injected connection) keep using the registered connection; that is also the minimal-connection-budget shape (no max_concurrency + 1 slot pool). See Jobs & Clients: SubJobEnqueuer.

Handle limitations¶

Handles returned by ctx.jobs.enqueue() do not have a client bound to them. Calling .status(), .refresh(), .attempts(), or .cancel() on these handles raises RuntimeError. .wait() works because it reads through the backend directly. To poll a sub-job's result from outside the actor body, pass its job_id to a full JobsClient instance:

sub_handle = await ctx.jobs.enqueue(process_item, ItemPayload(item_id=item_id))
job_id = sub_handle.job_id  # safe: job_id is always available

# Later, from application code with a full client:
result_handle = await client.get(job_id, result_adapter=process_item.result_adapter)

Progress reporting¶

Actors can emit structured progress updates that are buffered in memory, published to Redis in real time, and periodically flushed to Postgres. Callers can subscribe to these events via JobHandle.progress_stream() or the HTTP SSE endpoint; see Progress & Streaming.

`ctx.progress()`¶

async def progress(
    self,
    *,
    step: int | None = None,
    percent: float | None = None,
    detail: str | None = None,
    data: dict[str, object] | None = None,
) -> None: ...

All arguments are optional. Each call merges the supplied fields into the accumulated pending_state using last-writer-wins semantics. Fields not supplied in a call are left unchanged.

Parameter	Type	Description
`step`	`int \\| None`	Incremental step counter (e.g. items processed).
`percent`	`float \\| None`	Completion percentage in `[0.0, 100.0]`.
`detail`	`str \\| None`	Human-readable status message.
`data`	`dict[str, object] \\| None`	Arbitrary structured data. Must serialise to JSON with string dict keys (a non-`str` key raises `TypeError`).

Coalescing. Multiple ctx.progress() calls between periodic flush ticks are coalesced: only the latest value for each field is written to Postgres. Redis publishes are coalesced: at most one publish per job is in flight at a time and the final publish always lands, so SSE consumers see the latest state, not literally every call.

Sequence numbers. Each call increments a strictly monotone seq counter. State-change events on the same stream (the dispatch running transition, retries, snoozes, and terminal exits) consume the next value too, so seq is a total order over the job's whole event stream. SSE consumers use seq to detect duplicate or out-of-order delivery and to resume after reconnecting via Last-Event-ID.

ProgressTooLarge. Raises taskq.exceptions.ProgressTooLarge if the serialised data payload exceeds WorkerSettings.progress_data_max_bytes. Keep data small; use detail for human-readable strings and data only for structured metadata.

@actor(queue="media")
async def transcode_video(payload: TranscodePayload, ctx: JobContext[TranscodePayload]) -> None:
    segments = await split_into_segments(payload.url)
    total = len(segments)

    for i, segment in enumerate(segments):
        await transcode_segment(segment)
        await ctx.progress(
            step=i + 1,
            percent=round((i + 1) / total * 100, 1),
            detail=f"Transcoded segment {i + 1}/{total}",
        )

    await ctx.progress(percent=100.0, detail="Done")

No-op without Redis. When the worker is running without Redis (no redis extra installed or TASKQ_REDIS_URL not set), ctx.progress() silently returns without publishing. Progress state is still flushed to Postgres at the end of the job.

Testing actors without a database¶

Use InMemoryBackend and FakeClock to test actor behaviour in unit tests without Postgres. InMemoryBackend simulates the full enqueue-dispatch-execute cycle including unique_for dedup, singleton enforcement, and max_pending backpressure. Both come from the test-support namespace; see the Testing API reference for its full surface.

import pytest
from datetime import datetime, timezone
from pydantic import BaseModel
from taskq import actor
from taskq.client import JobsClient
from taskq.testing.in_memory import InMemoryBackend
from taskq.testing.clock import FakeClock


class MyPayload(BaseModel):
    value: int


class MyResult(BaseModel):
    doubled: int


@actor
async def double_value(payload: MyPayload) -> MyResult:
    return MyResult(doubled=payload.value * 2)


async def test_double_value():
    clock = FakeClock(start=datetime.now(timezone.utc))
    backend = InMemoryBackend(clock=clock)
    client = JobsClient(backend)

    # Register a stub so run_until_drained knows how to execute the actor.
    # Pass the ActorRef, not the bare name: the ref carries the declared
    # payload model, so the stub validates payloads against it (a bare name
    # falls back to the permissive PassthroughPayload and emits a warning).
    backend.register_stub(
        double_value,
        lambda payload, ctx: {"doubled": payload.value * 2},
    )

    handle = await client.enqueue(double_value, MyPayload(value=21))
    await backend.run_until_drained()
    result = await handle.wait()
    assert result.doubled == 42

The stub receives (payload: dict, ctx) where ctx is a minimal duck-typed object with job_id, attempt, payload, and cancel_event. For direct in-process invocation without the stub mechanism:

async def test_actor_direct():
    result = await double_value(MyPayload(value=21))
    assert result.doubled == 42

Direct invocation bypasses the queue entirely and is the simplest option when the actor has no DI dependencies and you do not need to test enqueue/dispatch behaviour.

JobsClient lifecycle. JobsClient is lightweight: it performs no I/O at construction. Create one instance per application and share it for the lifetime of the process. The connection pool is owned by the Backend, not the client. Creating a JobsClient per-request adds unnecessary overhead and does not provide isolation benefits.

Full worked example¶

An actor that ties together payload, result, context, a DI dependency, retry, and unique_for:

from datetime import timedelta
from pydantic import BaseModel
from taskq import actor
from taskq.context import JobContext
from taskq.exceptions import RetryAfter, Snooze
from taskq.retry import RetryPolicy

# --- Models ---


class OrderPayload(BaseModel):
    order_id: str
    customer_id: str
    amount_cents: int


class OrderResult(BaseModel):
    confirmation_number: str
    charged_at: str


# --- DI dependency (registered with the worker's DI container) ---


class PaymentGateway:
    async def charge(self, order_id: str, amount_cents: int) -> dict[str, str]: ...
    async def status(self, order_id: str) -> str: ...


# --- Actor ---


@actor(
    name="process_order",
    queue="payments",
    retry=RetryPolicy(kind="transient", max_attempts=5, base=timedelta(seconds=10)),
    result_ttl=timedelta(hours=24),
    unique_for=timedelta(minutes=10),
    unique_states=("pending", "scheduled", "running"),
    max_pending=5000,
)
async def process_order(
    payload: OrderPayload,
    ctx: JobContext[OrderPayload],
    *,
    gateway: PaymentGateway,
) -> OrderResult:
    ctx.log.info("charging", order_id=payload.order_id, attempt=ctx.attempt)

    if ctx.cancellation_requested:
        raise Snooze(timedelta(minutes=1))

    current_status = await gateway.status(payload.order_id)
    if current_status == "already_charged":
        return OrderResult(confirmation_number="DEDUP", charged_at="")

    try:
        charge = await gateway.charge(payload.order_id, payload.amount_cents)
    except RateLimitError as exc:
        raise RetryAfter(timedelta(seconds=exc.retry_after_seconds))

    # Enqueue a follow-up job for the receipt.
    await ctx.jobs.enqueue(
        send_receipt,
        ReceiptPayload(order_id=payload.order_id, email=payload.customer_id),
    )

    return OrderResult(
        confirmation_number=charge["confirmation"],
        charged_at=charge["charged_at"],
    )


# --- Enqueue ---


async def submit_order(client, order_id: str, customer_id: str, amount_cents: int):
    from taskq.exceptions import MaxPendingExceededError

    try:
        handle = await client.enqueue(
            process_order,
            OrderPayload(
                order_id=order_id,
                customer_id=customer_id,
                amount_cents=amount_cents,
            ),
            identity_key=order_id,  # required for unique_for dedup to take effect
        )
    except MaxPendingExceededError:
        raise RuntimeError("payment queue is full")

    if handle.was_existing:
        print(f"order {order_id} already queued: {handle.job_id}")
        return handle.job_id

    result: OrderResult = await handle.wait(timeout=30.0)
    print(f"confirmed: {result.confirmation_number}")
    return handle.job_id

A runnable companion: the examples/actors/ package organizes a toy fleet by feature domain (one module per domain — basic, failure, ratelimit, chained, DI, batch, advanced, cron, progress, tags, sync, real-world), each with a docstring naming the feature and what to observe; see its README and run the stack with docker compose up in examples/. For the production tour of the same engine, examples/fleet_demo/ narrates rate-limit denials, operator cancels, cron budgets, and SIGTERM deploys through a real worker's logs (uv run python -m examples.fleet_demo.run_demo).

Actor deregistration¶

Actors registered by worker startup create actor_config rows that persist until explicitly removed. For long-lived deployments this is intentional: the row is the source of truth for capacity and routing. For ephemeral, per-run deployments (e.g. my-actor.<run-id>), each run leaves a row behind.

`client.actors.deregister()`¶

from taskq import TaskQ

async with TaskQ(dsn=...) as tq:
    result = await tq.actors.deregister("my-actor.run-123")
    # force=False: refuses if non-terminal jobs or enabled schedules exist

Safety checks (force=False): - Refuses if any non-terminal jobs (pending/scheduled/running) reference the actor. - Refuses if any enabled cron schedules reference the actor.

force=True: - Still refuses if running jobs exist (they are actively executing). - Cancels pending/scheduled jobs (marks as cancelled with error_class='ActorDeregistered'). - Disables enabled cron schedules (sets enabled=false).

Terminal job history is never deleted. The jobs.actor column is plain text, not a foreign key, so terminal rows remain queryable by actor name after deregistration.

Queue cleanup (purge_queue=True): deletes the queues row if no other actor_config references the same queue. A shared queue is never purged.

Refusal types and return types¶

Every refusal subclasses ActorDeregistrationError (which itself subclasses TaskQError), so an except ActorDeregistrationError clause catches all three shapes; the full hierarchy renders on the Exceptions API reference.

Exception	Raised when	Machine-readable fields
`ActorHasActiveJobsError`	Non-terminal jobs reference the actor. With `force=True`, still raised when running jobs exist (they cannot be force-cancelled).	`actor`; `active_count` (int); `status_counts` (dict status → count) — decide "cancel them first" vs `force=True` from the fields, not the message.
`ActorHasEnabledSchedulesError`	Enabled cron schedules reference the actor.	`actor`; `schedule_ids` (list of schedule ids) — disable or delete them first, or pass `force=True` to disable them automatically.
`ActorNotFoundError`	No stored `actor_config` row exists — nothing to deregister (see Idempotent deregistration).	`actor`.

On success deregister() returns a DeregisterResult:

Field	Type	Meaning
`actor`	`str`	The deregistered actor's name.
`queue`	`str`	The actor's queue.
`actor_config_deleted`	`bool`	Always `True` — a missing row raises `ActorNotFoundError` instead; the field is retained for API-contract clarity and consumer assertions.
`schedules_disabled`	`int`	Enabled schedules disabled by this call (`force=True` only).
`jobs_cancelled`	`int`	Pending/scheduled jobs cancelled by this call (`force=True` only).
`terminal_jobs_remaining`	`int`	Terminal jobs left in place (history is never deleted).
`queue_purged`	`bool`	Whether the `queues` row was deleted (`purge_queue=True` and no other actor_config shares the queue).

The row snapshot form (ActorConfigRow: actor, max_concurrent, max_pending, queue, result_ttl, metadata, updated_at) is what the ActorsClient read methods (get, list) return; both types are defined in taskq.actor_config_ops.

Enqueue after deregistration¶

After deregistration, any client can still enqueue() the dead actor name; the INSERT succeeds (there is no foreign key from jobs.actor to actor_config.actor), and the job sits in pending status forever. Because the dispatch query inner-joins actor_config, the job will never be dispatched and no background sweep will reap it.

Operational discipline: stop enqueuing to an actor before deregistering it. Deregistration is best-effort against concurrent enqueue/dispatch; callers must quiesce the actor first.

Stop workers first: A concurrent worker startup (sync_actor_config) can re-create the actor_config row after deregistration, with capacity fields reset to @actor(...) defaults. Stop all workers for the actor before calling deregister.

Idempotent deregistration¶

A second deregister call on an already-deregistered actor raises ActorNotFoundError. For cleanup-automation loops:

from taskq.exceptions import ActorNotFoundError

try:
    await tq.actors.deregister(actor_name, force=True, purge_queue=True)
except ActorNotFoundError:
    pass  # already deregistered: idempotent

CLI¶

taskq actor-config deregister my-actor.run-123
taskq actor-config deregister my-actor.run-123 --force --purge-queue

Admin UI¶

The /admin/actors page lists all actor_config rows with active job counts and schedule counts. Each row has a deregister form with force and purge_queue checkboxes (requires TASKQ_ADMIN_ACTIONS_ENABLED=true). Each row also carries the executor statistics the shared per-actor stats read computes: completed jobs (live terminal rows included), failures with their share, the most recent error_class, and duration percentiles, window toggle for the recency view; see admin-ui.md: GET /admin/actors and the taskq.jobs.reclaimed{actor, disposition} counter for the per-actor crash split the expired-locks sweep records.

Actor API¶

Contents¶

The @actor decorator¶

Parameters¶

Choosing a concurrency control¶

Worked example: "I need a real mutex"¶

Decoration-time validation¶

Handler signatures¶

Payload only¶

Payload and context¶

Payload and DI dependencies¶

Payload, context, and DI dependencies¶

Sync actors¶

Cancellation for sync actors¶

Sync actors and shutdown¶

DI thread safety¶

Rate limiting and sub-jobs¶

ActorRef[P, R]¶

Properties¶

Direct invocation (__call__)¶

JobContext[P]¶

Fields¶

cancellation_requested property¶

Payload and result types¶

Singleton actors¶

unique_for deduplication¶

max_pending backpressure¶

Rate limits and reservations¶

Retry policy¶

RetryPolicy fields¶

Control-flow exceptions¶

Snooze¶

RetryAfter¶

Dependency injection¶

Registering providers¶

Scope lifetimes¶

Declaring dependencies in an actor¶

Sub-job enqueuing¶

Transaction semantics¶

Handle limitations¶

Progress reporting¶

ctx.progress()¶

Testing actors without a database¶

Full worked example¶

Actor deregistration¶

client.actors.deregister()¶

Refusal types and return types¶

Enqueue after deregistration¶

Idempotent deregistration¶

CLI¶

Admin UI¶

The `@actor` decorator¶

`ActorRef[P, R]`¶

Direct invocation (`call`)¶

`JobContext[P]`¶

`cancellation_requested` property¶

`unique_for` deduplication¶

`max_pending` backpressure¶

`RetryPolicy` fields¶

`Snooze`¶

`RetryAfter`¶

`ctx.progress()`¶

`client.actors.deregister()`¶