arq

Requires arq 0.26+, Python 3.11+. See the compatibility matrix for the full pin string.

Install

pip install z4j-arq

What it captures

arq doesn't expose signals. z4j chains worker on_job_start and after_job_end hooks; after_job_end reads arq's retained job result.

Source	z4j event
`on_job_start` hook	`task.started`
`after_job_end`, retained result succeeded	`task.succeeded`
`after_job_end`, retained result failed	`task.failed`
`after_job_end`, job queued or deferred again	`task.retried`

The adapter reports no queue or worker list (list_queues and list_workers return empty lists); the dashboard shows what the hook events describe.

Wiring

arq's config is a WorkerSettings class. attach_to_worker_settings chains z4j's on_job_start and after_job_end capture onto either a class (before arq builds the Worker) or an explicitly constructed Worker instance (before running it). Existing user start, end, and after-end hooks still run once:

from arq.connections import RedisSettings
from z4j_arq import ArqEngineAdapter, attach_to_worker_settings
from z4j_bare import install_agent

async def my_task(ctx):
    ...

class WorkerSettings:
    functions = [my_task]
    redis_settings = RedisSettings()
    keep_result = 3600  # completion capture reads the stored native result

adapter = ArqEngineAdapter(
    redis_settings=WorkerSettings.redis_settings,
    function_names=["my_task"],  # arq's registered name for this function
    queue_name="arq:queue",  # the default; match the worker's queue
)

# Chain z4j's hooks onto the class BEFORE arq instantiates the Worker.
attach_to_worker_settings(WorkerSettings, adapter=adapter)

# The hooks only capture; the worker process also needs the agent.
runtime = install_agent(engines=[adapter])

For a programmatically constructed worker, call attach_to_worker_settings(worker, adapter=adapter) before await worker.async_run(). Native arq does not provide ctx['worker'] in on_startup. For the arq CLI, attach to the settings class before worker construction. Repeated attachment replaces z4j's capture without nesting it.

Native arq does not put the function name or success flag in callback context. The adapter reads the job metadata at start and the result after arq commits it. Terminal capture therefore requires result retention, for example keep_result > 0 or keep_result_forever=True (including per-function overrides), and readable result serialization. With no retained result, the adapter logs that it cannot prove the outcome and omits the terminal event; it never reports an assumed success. This limitation also applies if native result serialization fails. A job still queued after an attempt is reported as retried.

arq cron jobs default to keep_result=0, which overrides worker-level retention. Set it explicitly for every cron entry whose completion you want to capture, for example cron(my_task, minute=0, keep_result=3600).

queue_name defaults to arq:queue and is the queue used for submission, get_task, and cancel. A per-submission queue overrides that default for the submitted job; reads and cancellation continue to use the adapter's configured queue. Events wait in an in-process queue of 10,000 entries; when it is full new events are dropped with a warning.

Actions

Verb	How
`submit`	`ArqRedis.enqueue_job(function_name, args, *kwargs)`; an ETA becomes `_defer_until` and a queue becomes `_queue_name`; a priority is refused
`cancel`	`Job.abort(timeout=10)`; success requires arq to confirm the abort; a timeout returns `failed` with `indeterminate: true`, because the abort request was written and may still land

Retry-by-id, bulk retry, purge, dead-letter listing and dead-letter requeue are not advertised; arq has no dead-letter store. A safe re-submission requires the function name and both complete argument collections; use submit_task with those explicit inputs.

Cron jobs

The arq-cron companion adapter inventories arq's code-defined cron jobs. Bound to the actual Worker, it also offers Run now; with NativeScheduleControl on that Worker, it applies dashboard create, edit, pause, resume and delete as durable overrides. See scheduler: arq-cron.

Caveats

No chord/group.
Task arguments are not read from arq's Redis job payload for lifecycle events.

FastAPI auto-wire

z4j_lifespan(arq_redis_settings=..., arq_function_names=..., arq_queue_name=...) registers the adapter in the FastAPI web process so it can submit and cancel, but that process runs no arq hooks. Capture happens only in the worker: construct the engine adapter with redis_settings there, then attach its hooks to the worker settings before arq builds the worker, as shown above.