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 uses worker on_job_start / on_job_end hooks.

Source	z4j event
`on_job_start` hook	`task.started`
`on_job_end` (success=True)	`task.succeeded`
`on_job_end` (success=False)	`task.failed`

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 on_job_end hooks onto either a class (before arq builds the Worker) or a Worker instance (from inside on_startup):

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()

adapter = ArqEngineAdapter(
    redis_settings=WorkerSettings.redis_settings,
    function_names=["myapp.worker.my_task"],
    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])

If you need to call it from inside on_startup instead (e.g. when the Worker is constructed via the arq CLI), pass ctx['worker'] as the target. The call is idempotent.

queue_name defaults to arq:queue and is the queue get_task and cancel look in. 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

arq's cron jobs are defined in WorkerSettings.cron_jobs. Same as Huey - code-only discovery, read-only. 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.