taskiq

Taskiq is an asynchronous distributed task queue for Python, built for teams who want Celery-style background processing without leaving async/await. It runs both sync and async functions, uses PEP-612 ParamSpec typing so kiq() calls are fully type-checked against the original task signature, and ships with pluggable brokers, result backends, serializers, and middlewares so the transport (Redis, NATS, RabbitMQ, Kafka, ZeroMQ, and more via the wider taskiq-* ecosystem) is decoupled from the task API itself.

Beyond simple fire-and-forget tasks, Taskiq bundles a scheduler with cron and interval support, a dependency-injection system (taskiq_dependencies) that lets you reuse FastAPI-style dependencies inside tasks, and a CLI (taskiq worker / taskiq scheduler) with filesystem task discovery and hot reload for local development. It’s a natural fit for teams already running FastAPI or AioHTTP who want their background job system to share the same async runtime, typing discipline, and dependency graph as their web layer instead of bolting on a synchronous queue like Celery.

What You Get

• A @broker.task decorator that turns any sync or async function into a fully typed, remotely-callable task via .kiq()
• A pluggable broker abstraction with official integrations for NATS, Redis, RabbitMQ, Kafka, and a built-in ZeroMQ broker
• A built-in scheduler (TaskiqScheduler) supporting cron expressions, fixed intervals, and one-off scheduled times
• A dependency-injection system shared with FastAPI-style dependencies, usable both in web handlers and background tasks
• Swappable serializers (JSON, ORJSON, MessagePack, CBOR, Pickle) and result backends for storing task output
• A CLI with worker/scheduler commands, filesystem task auto-discovery (--fs-discover), and hot reload for local dev
• Middleware hooks for retries (simple and smart backoff), Prometheus metrics, and OpenTelemetry tracing out of the box

Common Use Cases

• Offloading slow work (emails, report generation, image processing) from FastAPI or AioHTTP request handlers
• Running scheduled/cron jobs (data syncs, cleanup tasks, digest emails) alongside an async web application
• Building a microservice-style task pipeline where multiple worker processes consume from a shared Redis/NATS/RabbitMQ broker
• Sharing dependency-injected resources (DB sessions, HTTP clients) between web request handlers and background task functions
• Migrating a Celery-based async-unfriendly job system to a broker-agnostic, natively async equivalent

Under The Hood

Architecture Taskiq is organized around the abstract AsyncBroker class (taskiq/abc/broker.py), which owns task registration, the .task() decorator, event handlers (startup/shutdown for both client and worker/scheduler processes), and pluggable middlewares, serializer, formatter, and result_backend collaborators. Calling .kiq() on a decorated task routes through AsyncKicker (taskiq/kicker.py), which builds a BrokerMessage, applies the broker’s serializer/formatter, and hands it to the broker’s abstract kick() method; workers pull messages back out via the broker’s listen() async generator. Scheduling, CLI commands (taskiq/cli/worker, taskiq/cli/scheduler), and the dependency-injection layer are separated into their own subpackages, so the core client API stays broker-agnostic while concrete transports (taskiq-redis, taskiq-nats, etc.) live in separate sibling packages that implement the same abstract interfaces.

Tech Stack The project targets Python 3.10-3.14, is fully type-hinted (py.typed, strict mypy), and depends on aiohttp, anyio, pydantic (v1-v3 compatible), pycron, and its own lightweight taskiq_dependencies package for DI. Optional extras add msgpack, orjson, cbor2, pyzmq, prometheus_client, and OpenTelemetry instrumentation, so the base install stays lean and heavier serialization/observability dependencies are opt-in. Packaging uses hatchling with versioningit for version derivation from git tags, and the project is managed end-to-end with uv (dependency groups, tox-uv for multi-version test matrices).

Code Quality The tests/ tree mirrors the package layout (abc, api, brokers, cli, depends, formatters, middlewares, receiver, scheduler, serializers, opentelemetry) with dedicated pytest suites per subsystem, run under pytest-xdist for parallelism and pytest-cov for coverage, plus explicit tox environments for Python 3.10 through 3.13. CI runs black, ruff (a wide rule set including Bandit security checks, McCabe complexity, and pydocstyle), and strict mypy as separate matrix jobs via pre-commit, and a custom lightweight Error base class enforces that every exception declares its templated fields at class-definition time via dataclass_transform, catching malformed exceptions before runtime. Error handling favors typed, templated exception hierarchies (TaskiqError → BrokerError/ResultBackendError/etc.) over bare exceptions or silent failures.

What Makes It Unique Where most Python task queues (Celery, RQ) were designed sync-first and later grew async support, Taskiq is async-native from its AsyncBroker abstraction down, while still supporting sync task functions transparently. Its dependency-injection system is interoperable with FastAPI’s, letting the same Depends-style dependencies used in HTTP handlers run inside background tasks — a distinctive integration story documented explicitly for FastAPI and AioHTTP. The .kiq() call is fully generic over the original function’s ParamSpec and return type via PEP-612, giving callers autocomplete and type-checking on task arguments that most competing queues’ string-keyed task-name APIs cannot offer.