{"id":4795,"library":"taskiq","title":"Taskiq: Asynchronous Distributed Task Queue","description":"Taskiq is an asynchronous distributed task queue for Python, inspired by projects like Celery and Dramatiq. It supports both synchronous and asynchronous functions and integrates with popular async frameworks like FastAPI and AioHTTP. Taskiq is actively maintained with frequent releases, currently at version 0.12.1.","status":"active","version":"0.12.1","language":"en","source_language":"en","source_url":"https://github.com/taskiq-python/taskiq","tags":["task queue","async","distributed","background tasks","worker"],"install":[{"cmd":"pip install taskiq","lang":"bash","label":"Install core library"},{"cmd":"pip install taskiq-redis","lang":"bash","label":"Install Redis broker and result backend (recommended for production)"},{"cmd":"pip install \"taskiq[reload]\"","lang":"bash","label":"Install with reload capabilities for workers"}],"dependencies":[{"reason":"Taskiq requires Python 3.10 or higher (up to Python 3.13) to run.","package":"python","optional":false},{"reason":"Provides Redis broker and result backend, commonly used for distributed setups.","package":"taskiq-redis","optional":true},{"reason":"Provides RabbitMQ broker, another popular choice for production deployments.","package":"taskiq-aio-pika","optional":true},{"reason":"Provides NATS broker, recommended for production use cases.","package":"taskiq-nats","optional":true},{"reason":"Required for the `reload` extra, used for hot-reloading workers during development.","package":"watchdog","optional":true},{"reason":"Required for the `metrics` extra, used to expose Prometheus metrics.","package":"prometheus_client","optional":true}],"imports":[{"note":"Used for quick local development without external services.","symbol":"InMemoryBroker","correct":"from taskiq import InMemoryBroker"},{"note":"Used to schedule tasks.","symbol":"TaskiqScheduler","correct":"from taskiq import TaskiqScheduler"},{"note":"Middleware for integrating with the Taskiq Dashboard.","symbol":"TaskiqAdminMiddleware","correct":"from taskiq.middlewares.taskiq_admin_middleware import TaskiqAdminMiddleware"},{"note":"Example for importing a specific broker (e.g., Redis).","symbol":"RedisStreamBroker","correct":"from taskiq_redis import RedisStreamBroker"},{"note":"Example for importing a specific result backend (e.g., Redis).","symbol":"RedisAsyncResultBackend","correct":"from taskiq_redis import RedisAsyncResultBackend"}],"quickstart":{"code":"import asyncio\nfrom taskiq import InMemoryBroker\n\n# 1. Create a broker instance.\n# InMemoryBroker is for local development only. For production, use taskiq-redis, taskiq-aio-pika, etc.\nbroker = InMemoryBroker()\n\n# 2. Define a task using the broker's decorator.\n@broker.task\nasync def add_numbers(a: int, b: int) -> int:\n    print(f\"Executing add_numbers({a}, {b})\")\n    await asyncio.sleep(0.1) # Simulate some async work\n    return a + b\n\nasync def main():\n    # 3. Startup the broker. This is crucial for proper functioning.\n    await broker.startup()\n\n    # 4. Send the task to the broker.\n    task = await add_numbers.kiq(1, 2)\n\n    # 5. Wait for the result.\n    result = await task.wait_result(timeout=5)\n\n    if result.is_err:\n        print(f\"Task failed: {result.error}\")\n    else:\n        print(f\"Task result: {result.return_value}\")\n    \n    # 6. Shutdown the broker.\n    await broker.shutdown()\n\nif __name__ == \"__main__\":\n    asyncio.run(main())\n\n# To run with a worker (for distributed brokers, e.g., Redis):\n# 1. Save the above code as 'my_app.py'.\n# 2. Start an external broker (e.g., Redis).\n# 3. Run the worker from your terminal:\n#    taskiq worker my_app:broker\n# 4. Run the Python script (my_app.py) to send tasks.","lang":"python","description":"This quickstart demonstrates how to define and execute an asynchronous task using Taskiq's `InMemoryBroker`. It covers defining a broker, decorating a function as a task, sending the task using `.kiq()`, and retrieving its result with `.wait_result()`. For distributed usage, an external worker process would be started to consume tasks from a real broker (like Redis or RabbitMQ)."},"warnings":[{"fix":"Upgrade Python to 3.10 or higher. If unable to upgrade, pin Taskiq to `<0.12.0` in your `requirements.txt`.","message":"Support for Python 3.9 was dropped in Taskiq version 0.12.0. Users on Python 3.9 must either upgrade their Python version or stay on an older Taskiq release.","severity":"breaking","affected_versions":">=0.12.0"},{"fix":"For production, use a dedicated broker like `taskiq-redis`, `taskiq-aio-pika`, or `taskiq-nats`, which require separate installation and an external message queue service.","message":"The `InMemoryBroker` is designed for local development and testing only. It does not send messages over a network and cannot be used for distributed task execution in production environments.","severity":"gotcha","affected_versions":"All"},{"fix":"Always include `await broker.startup()` at the beginning and `await broker.shutdown()` at the end of your client-side application's lifecycle, typically within `asyncio.run()` or your main application entry point.","message":"Calling `broker.startup()` and `broker.shutdown()` is essential for all brokers. Failing to call `startup()` can lead to undefined behavior or tasks not being processed, while `shutdown()` ensures resources are properly released.","severity":"gotcha","affected_versions":"All"},{"fix":"Upgrade to Taskiq `0.12.1` or newer, where this issue has been resolved. If upgrading is not immediately possible, avoid using `dataclasses` with tasks monitored by `TaskiqAdminMiddleware` on `0.12.0`.","message":"In Taskiq `0.12.0`, `TaskiqAdminMiddleware` did not correctly handle tasks with `dataclasses` in their arguments or return values, leading to serialization issues.","severity":"gotcha","affected_versions":"0.12.0"},{"fix":"Specify a more unique `--tasks-pattern` (e.g., `my_tasks.py`) or explicitly list task modules: `taskiq worker your_app.broker:broker your_app.tasks.module_a your_app.tasks.module_b`.","message":"When using `taskiq worker --fs-discover` (file system discover) with the default `--tasks-pattern 'task.py'`, Taskiq may attempt to import `task.py` files from third-party libraries installed in your virtual environment, leading to `ImportError` or unexpected behavior.","severity":"gotcha","affected_versions":"All"},{"fix":"If using `ZeroMQBroker`, ensure you run only a single worker instance (e.g., `taskiq worker your_app:broker -w 1`). For multi-worker setups, choose a different broker like Redis or RabbitMQ.","message":"The `ZeroMQBroker` is explicitly stated to be suitable for projects with only ONE worker process. If multiple workers are connected to a `ZeroMQBroker`, tasks will be executed N times (where N is the number of workers) due to ZMQ's publish-subscribe architecture.","severity":"gotcha","affected_versions":"All"}],"env_vars":null,"last_verified":"2026-04-12T00:00:00.000Z","next_check":"2026-07-11T00:00:00.000Z"}