taskiq

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

Install

pip install z4j-taskiq

Wiring

taskiq's extensibility is middleware-based. The framework adapters wire it automatically; for bare usage, use attach_to_broker or add the middleware manually:

from taskiq_redis import RedisAsyncResultBackend, ListQueueBroker
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)

attach_to_broker adds Z4JTaskiqMiddleware and is the supported entry point. 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`
terminal `on_error`	`task.failed`

Actions

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

Retry, cancel, bulk retry, and purge 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 but each process is a separate agent.
AioPikaBroker (RabbitMQ) - queue length via RabbitMQ management API.
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.