Events & Logs

Events & webhooks

queue.on_event()

queue.on_event(event: str) -> Callable

Register a callback for a queue event. Supported events: job.completed, job.failed, job.retried, job.dead.

@queue.on_event("job.failed")
def handle_failure(job_id: str, task_name: str, error: str) -> None:
    ...

queue.add_webhook()

queue.add_webhook(
    url: str,
    events: list[EventType] | None = None,
    headers: dict[str, str] | None = None,
    secret: str | None = None,
    max_retries: int = 3,
    timeout: float = 10.0,
    retry_backoff: float = 2.0,
) -> None

Register a webhook URL for one or more events. 4xx responses are not retried; 5xx responses are retried with exponential backoff.

Logs

Durable, per-job structured log rows in shared storage — not process stdout, so a producer or dashboard reads what a worker on another host wrote. Beyond diagnostics they carry streaming partial results: a running task publishes at level "result", and those rows are what a stream consumer polls.

queue.task_logs()

queue.task_logs(job_id: str) -> list[dict]

Every row for one job, oldest first, published partials included. Each dict carries id, job_id, task_name, level, message, extra and logged_at (Unix milliseconds). Declared as get_task_logs.

queue.query_logs()

queue.query_logs(
    task_name: str | None = None,
    level: LogLevel | None = None,
    since: int = 3600,
    limit: int = 100,
) -> list[dict]

The same rows across jobs, newest first. since is an age in seconds (default one hour), not a timestamp, and limit defaults to 100 — both exist so an unfiltered call cannot scan the whole table. level is a LogLevel, not a bare string. Declared as query_task_logs, whose since_seconds this passes through.

Writing is the running task's job, not the queue's: current_job.log() and current_job.publish() are the API, and write_task_log is what they call — which is why it takes the job_id and task_name the context already knows.