Taskiq: Asynchronous Distributed Task Queue

Warnings

• breaking 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.
  Fix: Upgrade Python to 3.10 or higher. If unable to upgrade, pin Taskiq to `<0.12.0` in your `requirements.txt`.
• gotcha 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.
  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.
• gotcha 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.
  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.
• gotcha In Taskiq `0.12.0`, `TaskiqAdminMiddleware` did not correctly handle tasks with `dataclasses` in their arguments or return values, leading to serialization issues.
  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`.
• gotcha 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.
  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`.
• gotcha 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.
  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.

Install

• pip install taskiq Install core library
• pip install taskiq-redis Install Redis broker and result backend (recommended for production)
• pip install "taskiq[reload]" Install with reload capabilities for workers

Imports

• InMemoryBroker

  from taskiq import InMemoryBroker

  Used for quick local development without external services.
• TaskiqScheduler

  from taskiq import TaskiqScheduler

  Used to schedule tasks.
• TaskiqAdminMiddleware

  from taskiq.middlewares.taskiq_admin_middleware import TaskiqAdminMiddleware

  Middleware for integrating with the Taskiq Dashboard.
• RedisStreamBroker

  from taskiq_redis import RedisStreamBroker

  Example for importing a specific broker (e.g., Redis).
• RedisAsyncResultBackend

  from taskiq_redis import RedisAsyncResultBackend

  Example for importing a specific result backend (e.g., Redis).

Quickstart

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

import asyncio
from taskiq import InMemoryBroker

# 1. Create a broker instance.
# InMemoryBroker is for local development only. For production, use taskiq-redis, taskiq-aio-pika, etc.
broker = InMemoryBroker()

# 2. Define a task using the broker's decorator.
@broker.task
async def add_numbers(a: int, b: int) -> int:
    print(f"Executing add_numbers({a}, {b})")
    await asyncio.sleep(0.1) # Simulate some async work
    return a + b

async def main():
    # 3. Startup the broker. This is crucial for proper functioning.
    await broker.startup()

    # 4. Send the task to the broker.
    task = await add_numbers.kiq(1, 2)

    # 5. Wait for the result.
    result = await task.wait_result(timeout=5)

    if result.is_err:
        print(f"Task failed: {result.error}")
    else:
        print(f"Task result: {result.return_value}")
    
    # 6. Shutdown the broker.
    await broker.shutdown()

if __name__ == "__main__":
    asyncio.run(main())

# To run with a worker (for distributed brokers, e.g., Redis):
# 1. Save the above code as 'my_app.py'.
# 2. Start an external broker (e.g., Redis).
# 3. Run the worker from your terminal:
#    taskiq worker my_app:broker
# 4. Run the Python script (my_app.py) to send tasks.