Workers & Hooks

Methods for running workers, lifecycle hooks, circuit breakers, and the sync/async method mapping.

Workers

queue.run_worker()

queue.run_worker(
    queues: Sequence[str] | None = None,
    tags: list[str] | None = None,
    pool: str = "thread",
    app: str | None = None,
) -> None

Start the worker loop. Blocks until interrupted.

queue.shutdown()

queue.shutdown() -> None

Request graceful worker shutdown — the programmatic equivalent of SIGINT/SIGTERM. The scheduler stops dispatching and run_worker() returns once running tasks finish (bounded by drain_timeout). Non-blocking and safe from any thread, including while a worker is still starting — the request is held until the runs it was aimed at have finished, rather than discarded. One queue can run several workers at once, each over its own queue list; one call stops all of them, and the request clears only once the last has exited. Declared as request_shutdown.

A signal does one thing more than this method: the handler first flips the worker's registry row to "draining" (declared as set_worker_status), so workers() shows the drain from another process. shutdown() skips that step — the worker still drains, it just doesn't announce it.

queue.workers()

queue.workers() -> list[dict]

Return live state of all registered workers, read from storage — so a producer or dashboard process sees the whole fleet, not just its own worker. Declared as list_workers.

The rows are written by each worker's own heartbeat (declared as worker_heartbeat), which beats every 5 seconds carrying that worker's resource-health snapshot. The same call doubles as the dead-worker reaper: it returns the ids it evicted, which is what turns a worker that stopped beating into a WORKER_OFFLINE event. So a stale last_heartbeat means the process is gone or wedged, and resource_health is only as fresh as the last beat.

Each dict contains:

await queue.aworkers()

await queue.aworkers() -> list[dict]

Async version of workers().

queue.arun_worker()

await queue.arun_worker(
    queues: Sequence[str] | None = None,
    tags: list[str] | None = None,
    pool: str = "thread",
    app: str | None = None,
) -> None

Async version of run_worker(). Runs the blocking worker loop in a thread executor so it does not block the asyncio event loop. Accepts the same pool and app kwargs as the sync variant (app is required when pool="prefork").

Circuit breakers

queue.circuit_breakers()

queue.circuit_breakers() -> list[dict]

Live state of every task that has a circuit breaker configured: task_name, state (closed / open / half_open), failure_count, last_failure_at, opened_at, and the breaker's own threshold, window_ms and cooldown_ms. Read from storage, so the state is the cluster's rather than this process's — a breaker one worker opened is open for all of them. Declared as list_circuit_breakers.

An open breaker with an opened_at older than cooldown_ms is one probe away from half_open; that pair is what tells you whether a task is still being shed or is already recovering.

Hooks

@queue.before_task

@queue.before_task
def hook(task_name: str, args: tuple, kwargs: dict) -> None: ...

@queue.after_task

@queue.after_task
def hook(task_name: str, args: tuple, kwargs: dict, result: Any, error: Exception | None) -> None: ...

@queue.on_success

@queue.on_success
def hook(task_name: str, args: tuple, kwargs: dict, result: Any) -> None: ...

@queue.on_failure

@queue.on_failure
def hook(task_name: str, args: tuple, kwargs: dict, error: Exception) -> None: ...

Async methods