DOCSEARCHbyAlgolia
Submit Docs

taskiq Documentation

repository·master·Indexed 25 days ago

https://github.com/taskiq-python/taskiq

An asynchronous distributed task queue for Python designed for asyncio workflows. Taskiq supports both synchronous and asynchronous functions and integrates with frameworks like FastAPI and AioHTTP. It features a modular architecture with extensible brokers, middlewares, result backends, and schedule sources, supporting backends such as RabbitMQ, Redis, NATS, and PostgreSQL.

Tokens
29.5K
Snippets
73
Records
165
Agent score
79%

What's inside taskiq

  1. What is taskiq?

    master

    Taskiq is a distributed task queue library for Python designed to send and process functions across different servers. It is specifically built to handle asynchronous functions using distributed queues (like RabbitMQ).

    Key characteristics include:

    • Modular Design: The core library is lightweight, providing basic functionality, a CLI, and abstractions for extension.
    • Extensibility: It is designed to be easily extended via various broker and result backend libraries.
    • Async-First: Unlike Celery or Dramatiq which are suited for synchronous projects, Taskiq is optimized for asyncio workflows.
    • Advanced Features: Supports dependency injection, global middlewares, multiple broker/result backends, startup/shutdown events, and task pipelines.

  2. Explore officially supported taskiq components

    master

    Taskiq provides a modular architecture through various officially supported plugins. You can find documentation for specific component types in the following sections:

    • Brokers: For managing task queues and task execution.
    • Middlewares: For intercepting task execution (e.g., for logging, metrics, or tracing).
    • Result Backends: For storing and retrieving task results.
    • Schedule Sources: For managing periodic task scheduling.

  3. Broker implementation conventions for delay and priority

    master

    While not strictly required, following these conventions ensures better compatibility with taskiq features:

    1. Delay: If a message contains a delay label (an int or float), the broker should delay the task execution by that number of seconds.
    2. Priority: If a message contains a priority label, the broker should handle the message with priority, ensuring tasks with higher priorities are executed sooner.

  4. How to extend the taskiq CLI with new subcommands

    master

    To add a new subcommand to the taskiq CLI, you must create a class that implements the taskiq.abc.cmd.TaskiqCMD abstract class.

    Inside the exec method of your class, you are responsible for parsing incoming arguments. Note that because all CLI arguments to taskiq are shifted, you can ignore the args parameter provided to the exec method. You can use external libraries like click or typer within your implementation to handle argument parsing and CLI structure.

  5. Configure sync function execution in workers

    master

    Taskiq executes synchronous functions in a separate thread or process. By default, it uses a ThreadPoolExecutor (suitable for IO-bound tasks). For CPU-intensive workloads (like neural network training), you should switch to a ProcessPoolExecutor.

    Use these CLI options to adjust behavior:

    • --use-process-pool: Switch to ProcessPoolExecutor.
    • --max-process-pool-processes: Manually specify the number of worker processes.
    • --max-threadpool-threads: Configure the maximum threads for ThreadPoolExecutor (if not using process pool).

  6. How taskiq works: Brokers and Tasks

    master

    Taskiq is an asynchronous distributed task queue. The core workflow involves two main components:

    1. Broker: An object that communicates with workers using distributed queues (e.g., NATS, Redis, RabbitMQ, Kafka). You must instantiate a broker specific to your backend.
    2. Tasks: Functions decorated with @broker.task.

    To run a task, you call the .kiq() method on the task function. The message is sent to the broker, which then routes it to a worker for execution. Always ensure you call await broker.startup() before sending tasks and await broker.shutdown() when cleaning up.

    import asyncio
from taskiq_nats import JetStreamBroker

broker = JetStreamBroker("nats://localhost:4222", queue="my_queue")

@broker.task
async def my_task(a: int, b: int) -> None:
    print("AB", a + b)

async def main():
    await broker.startup()
    await my_task.kiq(1, 2)
    await broker.shutdown()

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

  7. Use multiple schedule sources and merge functions

    master

    The TaskiqScheduler can combine schedules from multiple sources. When merging schedules from different sources, you can provide a custom merge function to resolve conflicts or apply complex logic (like filtering).

    Taskiq provides two default merge functions in the taskiq.scheduler.merge_functions module:

    • preserve_all: Simply adds new schedules to the existing ones.
    • only_unique: Adds a schedule only if it hasn't been added by a previous source.

  8. What to expect from taskiq framework integrations

    master

    Taskiq provides integrations for various web frameworks to simplify development. These integrations primarily provide two capabilities to your task handlers:

    1. Lifecycle Management: Automatic handling of Startup and Shutdown events for the framework.
    2. Dependency Injection: Integration of framework-specific Dependencies that can be used directly within your task handlers.

  10. How Taskiq architecture works

    master

    Taskiq follows a distributed architecture split between the client side and the worker side:

    1. Client Side: Uses kickers to assemble messages and brokers to send them to an external system (like Redis).
    2. Worker Side: The broker receives messages from the external system, executes the task, and saves the results in a result backend.
    3. Result Retrieval: The client can then retrieve the results from the result backend.

    Core Components:

    • Broker: The central component. It must implement AsyncBroker (specifically kick to send and listen to receive).
    • Kicker: An object used to form messages. It allows customizing the broker, adding labels, or changing the task_id before sending.
    • Messages: Data packets containing the task name, arguments, and labels.
    • Result Backend: Stores and retrieves task results (of type TaskiqResult). Brokers use an AsyncResultBackend implementation.
    • Workers: Processes that run the broker.listen() loop to execute tasks.
    • Middlewares: Interceptors that can modify messages or perform actions during the task lifecycle.
    • Context: Provides access to task metadata and execution control (like requeueing) from within the task itself.

  11. Manage global variables with TaskiqState

    master

    The TaskiqState is a global variable used to store objects that you want to persist across the lifecycle of a worker or client, such as database connection pools. You can populate or clean up this state by attaching event handlers to specific lifecycle events.

    Available events:

    • WORKER_STARTUP: Called when the worker starts listening to broker messages.
    • CLIENT_STARTUP: Called when the startup method of your broker is called.
    • WORKER_SHUTDOWN: Called when the worker shuts down.
    • CLIENT_SHUTDOWN: Called when the client shuts down.

    You can add handlers using the @broker.on_event(event) decorator or programmatically via broker.add_event_handler.

    @broker.on_event(TaskiqEvents.WORKER_STARTUP)
async def startup(state: TaskiqState) -> None:
    # Store connection pool on startup for later use
    state.redis = ConnectionPool.from_url("redis://localhost/1")

  12. Resolve FastAPI dependencies in Taskiq tasks using TaskiqDepends

    master

    When using FastAPI dependencies that rely on fastapi.Request or fastapi.HTTPConnection, Taskiq cannot resolve them automatically. You must explicitly mark these parameters with TaskiqDepends so the dependency injection system knows to resolve them.

    Important: The Request or HTTPConnection object injected into a task is a mocked version and is not the same request/connection that was active when the task was originally sent.

    Use either Annotated (Python 3.10+) or default values to apply TaskiqDepends.

    # Using Annotated (Recommended for 3.10+)
from typing import Annotated
from fastapi import Request
from taskiq import TaskiqDepends

async def get_redis_pool(request: Annotated[Request, TaskiqDepends()]):
    return request.app.state.redis_pool

# Using default values
from fastapi import Request
from taskiq import TaskiqDepends

async def get_redis_pool(request: Request = TaskiqDepends()):
    return request.app.state.redis_pool