taskiq

Requires TaskIQ 0.11+, Python 3.11+. See the compatibility matrix for the full pin string.

Install

pip install z4j-taskiq

Wiring

taskiq's extensibility is middleware-based. Flask wires it from app.config["TASKIQ_BROKER"] and FastAPI from z4j_lifespan(taskiq_broker=...); z4j-django discovers only Celery. For Django and bare usage, use attach_to_broker and start the agent yourself:

from taskiq_redis import RedisAsyncResultBackend, ListQueueBroker
from z4j_bare import install_agent
from z4j_taskiq import TaskiqEngineAdapter, attach_to_broker

broker = ListQueueBroker("redis://localhost:6379")
broker = broker.with_result_backend(RedisAsyncResultBackend("redis://..."))
adapter = TaskiqEngineAdapter(broker=broker)
attach_to_broker(broker, adapter=adapter)
runtime = install_agent(engines=[adapter])

attach_to_broker adds Z4JTaskiqMiddleware and is the supported entry point. The middleware only captures; a bare worker also needs install_agent(engines=[adapter]) to run the agent that drains and ships the events. Events wait in an in-process queue of 10,000 entries, and when it is full new events are dropped with a warning. Attach it before the TaskIQ CLI or your application lifespan starts the broker; that host owns startup and shutdown, so do not start an already host-managed broker again. A standalone manual client must start and shut down its broker exactly once in its own lifecycle. The middleware startup hook records the TaskIQ broker's owner event loop; z4j then marshals submit and result-backend reconciliation from the agent's background loop back to that owner loop.

When installing from a TaskIQ startup callback without the middleware, bind the same live loop explicitly:

import asyncio

adapter = TaskiqEngineAdapter(
    broker=broker,
    broker_loop=asyncio.get_running_loop(),
)

Do not obtain broker_loop from a temporary asyncio.run() call because that loop is closed as soon as the call returns. Before startup binding or explicit binding, asynchronous submit and result-backend probes fail closed rather than guessing that the agent's background loop owns the broker.

The binding is released during TaskIQ shutdown and may be established again by a later broker lifecycle. A broker accepts only one z4j middleware/adapter pair. Capture failures inside z4j hooks are logged by exception type and the event is dropped; they do not abort TaskIQ enqueue or task execution.

What it captures

taskiq hook	z4j event
`pre_send`	`task.received` or `task.retried`
`pre_execute`	`task.started`
`post_execute` (no exception)	`task.succeeded`
`on_error` when the retry middleware will re-enqueue and `pre_send` did not already record it	`task.retried`
terminal `on_error`	`task.failed`

Actions

Verb	How
`submit`	registered task's `.kiq(args, *kwargs)`

Retry, cancel, bulk retry, purge, dead-letter listing and dead-letter requeue are not advertised because TaskIQ's broker-independent layer has no portable primitive for them. Submit accepts an omitted queue or z4j's canonical logical default queue; both use the broker's configured default. A non-default queue, ETA, or priority is rejected before enqueue because those overrides cannot be honored portably.

Multi-broker

TaskIQ itself can use multiple brokers, but one z4j agent runtime can register only one taskiq engine entry. Every TaskiqEngineAdapter has the same engine name, and the runtime stores adapters by that name, so a later adapter would replace the earlier one. Run separate agent processes, with separate agent credentials when you want separate dashboard agents, for multiple brokers.

Schedules

The standard taskiq label schedule source is read-only inventory. See scheduler: taskiq-scheduler.

Caveats

ZeroMQBroker - in-memory distribution between processes. Works with z4j; processes that share one agent token appear as workers under one agent, each with its own worker_id.
AioPikaBroker (RabbitMQ) - lifecycle capture only.
The adapter reports no queue depth or worker list for any broker (list_queues and list_workers return empty lists).
InMemoryBroker - local-only, useful for tests; z4j captures events but there's no cross-process visibility.

Multi-broker config

Attach the broker used by this runtime with attach_to_broker(broker, adapter=adapter). A process that needs another TaskIQ broker should run a separate z4j runtime rather than attaching a second same-name adapter to the first runtime.