Taskiq

https://taskiq-python.github.io
Admin
Taskiq is an async distributed task manager for Python that provides production-ready task queue...
Tokens:35,050
Snippets:338
Trust Score:8.3
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install dependencies with uv sync

Source: https://taskiq-python.github.io/contrib.html

Installs all dependencies for the taskiq repository using uv. Run this after cloning the repo and installing uv.

```bash
uv sync --all-extras
```

--------------------------------

### Install taskiq-aiohttp

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-aiohttp.html

Install the taskiq-aiohttp library to integrate Taskiq with AioHTTP.

```bash
pip install "taskiq-aiohttp"
```

--------------------------------

### Install Prometheus client

Source: https://taskiq-python.github.io/available-components/middlewares.html

Install the prometheus_client package to enable Prometheus metrics.

```bash
pip install "prometheus_client"
```

--------------------------------

### Install taskiq-faststream

Source: https://taskiq-python.github.io/framework_integrations/faststream.html

Install the taskiq-faststream library to use FastStream as a broker for Taskiq.

```bash
pip install "taskiq-faststream"
```

--------------------------------

### Install taskiq-aiogram

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-aiogram.html

Install the taskiq-aiogram library to enable the integration.

```bash
pip install "taskiq-aiogram"
```

--------------------------------

### Install Taskiq OpenTelemetry extras

Source: https://taskiq-python.github.io/available-components/middlewares.html

Install Taskiq with OpenTelemetry support.

```bash
pip install "taskiq[opentelemetry]"
```

--------------------------------

### Run Scheduler with Example Arguments

Source: https://taskiq-python.github.io/guide/cli.html

Example command that specifies the scheduler location and additional modules to import.

```bash
taskiq scheduler my_project.broker:scheduler my_project.module1 my_project.module2
```

--------------------------------

### Install taskiq from Git

Source: https://taskiq-python.github.io/guide/getting-started.html

Install the latest development version of taskiq directly from the GitHub repository.

```bash
pip install git+https://github.com/taskiq-python/taskiq.git
```

--------------------------------

### Use taskiq-aiogram in bot.py

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-aiogram.html

Example bot.py that sets up the aiogram bot and dispatcher, starts and stops the taskiq broker on startup/shutdown (only in the client process to avoid infinite loops), and handles a /task command that enqueues the my_task task.

```python
import asyncio
import logging
import sys

from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command

from tkq import broker, my_task

dp = Dispatcher()
bot = Bot(token="TOKEN")


# Taskiq calls this function when starting the worker.
@dp.startup()
async def setup_taskiq(bot: Bot, *_args, **_kwargs):
    # Here we check if it's a client-side,
    # Because otherwise you're going to
    # create infinite loop of startup events.
    if not broker.is_worker_process:
        logging.info("Setting up taskiq")
        await broker.startup()


# Taskiq calls this function when shutting down the worker.
@dp.shutdown()
async def shutdown_taskiq(bot: Bot, *_args, **_kwargs):
    if not broker.is_worker_process:
        logging.info("Shutting down taskiq")
        await broker.shutdown()


## Simple command to handle
@dp.message(Command("task"))
async def message(message: types.Message):
    await my_task.kiq(message.chat.id)


## Main function that starts the bot.
async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, stream=sys.stdout)
    asyncio.run(main())
```

--------------------------------

### Install taskiq-postgres

Source: https://taskiq-python.github.io/available-components/schedule-sources.html

Installation command for the PostgreSQL schedule source package.

```bash
pip install taskiq-postgres
```

--------------------------------

### Using TaskiqState with default values syntax

Source: https://taskiq-python.github.io/guide/state-and-deps.html

Same example as the Annotated version but using default values for dependency injection: context: Context = TaskiqDepends(). Requires installing taskiq, taskiq-redis, and taskiq-aio-pika. The worker initializes the Redis pool at startup and closes it on shutdown.

```python
import asyncio

from redis.asyncio import ConnectionPool, Redis  # type: ignore
from taskiq_aio_pika import AioPikaBroker
from taskiq_redis import RedisAsyncResultBackend

from taskiq import Context, TaskiqDepends, TaskiqEvents, TaskiqState

# To run this example, please install:
# * taskiq
# * taskiq-redis
# * taskiq-aio-pika

broker = AioPikaBroker(
    "amqp://localhost",
).with_result_backend(RedisAsyncResultBackend("redis://localhost"))


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


@broker.on_event(TaskiqEvents.WORKER_SHUTDOWN)
async def shutdown(state: TaskiqState) -> None:
    # Here we close our pool on shutdown event.
    await state.redis.disconnect()


@broker.task
async def get_val(key: str, context: Context = TaskiqDepends()) -> str | None:
    # Now we can use our pool.
    redis = Redis(connection_pool=context.state.redis, decode_responses=True)
    return await redis.get(key)


@broker.task
async def set_val(key: str, value: str, context: Context = TaskiqDepends()) -> None:
    # Now we can use our pool to set value.
    await Redis(connection_pool=context.state.redis).set(key, value)


async def main() -> None:
    await broker.startup()

    set_task = await set_val.kiq("key", "value")
    set_result = await set_task.wait_result(with_logs=True)
    if set_result.is_err:
        print(set_result.log)
        raise ValueError("Cannot set value in redis. See logs.")

    get_task = await get_val.kiq("key")
    get_res = await get_task.wait_result()
    print(f"Got redis value: {get_res.return_value}")

    await broker.shutdown()


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

--------------------------------

### Install taskiq with CBOR serializer

Source: https://taskiq-python.github.io/guide/message-format.html

Install taskiq with the CBOR serializer extra.

```bash
pip install "taskiq[cbor]"
```

--------------------------------

### Install taskiq with MSGPack serializer

Source: https://taskiq-python.github.io/guide/message-format.html

Install taskiq with the MSGPack serializer extra.

```bash
pip install "taskiq[msgpack]"
```

--------------------------------

### Install S3 result backend

Source: https://taskiq-python.github.io/available-components/result-backends.html

Install the taskiq-aio-sqs package to use the S3 result backend.

```bash
pip install taskiq-aio-sqs
```

--------------------------------

### Install Taskiq with zmq extra

Source: https://taskiq-python.github.io/available-components/brokers.html

Install Taskiq with the zmq extra to use ZeroMQBroker. This installs pyzmq as a dependency.

```bash
pip install "taskiq[zmq]"
```

--------------------------------

### Using TaskiqState with Annotated 3.10+ syntax

Source: https://taskiq-python.github.io/guide/state-and-deps.html

Example of storing a Redis connection pool in TaskiqState during worker startup and accessing it in tasks using Annotated[Context, TaskiqDepends()]. Requires installing taskiq, taskiq-redis, and taskiq-aio-pika. The worker initializes the Redis pool at startup and closes it on shutdown.

```python
import asyncio
from typing import Annotated

from redis.asyncio import ConnectionPool, Redis  # type: ignore
from taskiq_aio_pika import AioPikaBroker
from taskiq_redis import RedisAsyncResultBackend

from taskiq import Context, TaskiqDepends, TaskiqEvents, TaskiqState

# To run this example, please install:
# * taskiq
# * taskiq-redis
# * taskiq-aio-pika

broker = AioPikaBroker(
    "amqp://localhost",
).with_result_backend(RedisAsyncResultBackend("redis://localhost"))


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


@broker.on_event(TaskiqEvents.WORKER_SHUTDOWN)
async def shutdown(state: TaskiqState) -> None:
    # Here we close our pool on shutdown event.
    await state.redis.disconnect()


@broker.task
async def get_val(
    key: str,
    context: Annotated[Context, TaskiqDepends()],
) -> str | None:
    # Now we can use our pool.
    redis = Redis(connection_pool=context.state.redis, decode_responses=True)
    return await redis.get(key)


@broker.task
async def set_val(
    key: str,
    value: str,
    context: Annotated[Context, TaskiqDepends()],
) -> None:
    # Now we can use our pool to set value.
    await Redis(connection_pool=context.state.redis).set(key, value)


async def main() -> None:
    await broker.startup()

    set_task = await set_val.kiq("key", "value")
    set_result = await set_task.wait_result(with_logs=True)
    if set_result.is_err:
        print(set_result.log)
        raise ValueError("Cannot set value in redis. See logs.")

    get_task = await get_val.kiq("key")
    get_res = await get_task.wait_result()
    print(f"Got redis value: {get_res.return_value}")

    await broker.shutdown()


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

--------------------------------

### Sync middleware example

Source: https://taskiq-python.github.io/extending-taskiq/middleware.html

Sync middleware example. All methods are sync and use time.sleep. Implements startup, shutdown, pre_execute, and post_save hooks.

```python
from time import sleep
from typing import Any

from taskiq import TaskiqMessage, TaskiqMiddleware, TaskiqResult


class MyMiddleware(TaskiqMiddleware):
    def startup(self) -> None:
        print("RUN STARTUP")
        sleep(1)

    def shutdown(self) -> None:
        print("RUN SHUTDOWN")
        sleep(1)

    def pre_execute(self, message: "TaskiqMessage") -> TaskiqMessage:
        sleep(1)
        return message

    def post_save(self, message: "TaskiqMessage", result: "TaskiqResult[Any]") -> None:
        sleep(1)
        print("Post save")
```

--------------------------------

### Install Taskiq metrics extras

Source: https://taskiq-python.github.io/available-components/middlewares.html

Install Taskiq with metrics extras to enable Prometheus support.

```bash
pip install "taskiq[metrics]"
```

--------------------------------

### Install pyzmq for ZeroMQBroker

Source: https://taskiq-python.github.io/available-components/brokers.html

Install the pyzmq library to use ZeroMQBroker. Required for the ZeroMQ broker.

```bash
pip install pyzmq
```

--------------------------------

### Install taskiq with ORJSON serializer

Source: https://taskiq-python.github.io/guide/message-format.html

Install taskiq with the ORJSON serializer extra.

```bash
pip install "taskiq[orjson]"
```

--------------------------------

### Example output of running broker.py

Source: https://taskiq-python.github.io/guide/getting-started.html

Terminal output when running the example broker.py. Shows the execution time and the returned value.

```bash
❯ python broker.py
Task execution took: 7.3909759521484375e-06 seconds.
Returned value: 2
```

--------------------------------

### Install taskiq-aio-pika

Source: https://taskiq-python.github.io/available-components/brokers.html

Install the AioPikaBroker for RabbitMQ. This broker is maintained by Taskiq developers and is not part of the core library.

```bash
pip install taskiq-aio-pika
```

--------------------------------

### Install Redis result backend

Source: https://taskiq-python.github.io/available-components/result-backends.html

Install the taskiq-redis package to use the Redis result backend.

```bash
pip install taskiq-redis
```

--------------------------------

### Define task and run with startup

Source: https://taskiq-python.github.io/guide/getting-started.html

Complete example showing how to define a task with @broker.task, call startup before sending tasks, use .kiq to send, wait for the result, and shutdown. The startup method is required to avoid undefined behavior.

```python
# broker.py
import asyncio

from taskiq import InMemoryBroker

broker = InMemoryBroker()


@broker.task
async def add_one(value: int) -> int:
    return value + 1


async def main() -> None:
    # Never forget to call startup in the beginning.
    await broker.startup()
    # Send the task to the broker.
    task = await add_one.kiq(1)
    # Wait for the result.
    result = await task.wait_result(timeout=2)
    print(f"Task execution took: {result.execution_time} seconds.")
    if not result.is_err:
        print(f"Returned value: {result.return_value}")
    else:
        print("Error found while executing task.")
    await broker.shutdown()


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

--------------------------------

### Install documentation dependencies with pnpm

Source: https://taskiq-python.github.io/contrib.html

Installs documentation dependencies. pnpm is recommended, but the package.json is compatible with npm and bun.

```bash
pnpm i
```

--------------------------------

### Install YDB result backend

Source: https://taskiq-python.github.io/available-components/result-backends.html

Install the taskiq-ydb package to use the YDB result backend.

```bash
pip install taskiq-ydb
```

--------------------------------

### Start the scheduler with taskiq scheduler command

Source: https://taskiq-python.github.io/guide/scheduling-tasks.html

Starts the scheduler by importing the scheduler object from the specified module. Only one scheduler instance should be run at a time to avoid duplicate task executions.

```bash
taskiq scheduler module:scheduler

```

--------------------------------

### Run documentation dev server with pnpm docs:dev

Source: https://taskiq-python.github.io/contrib.html

Starts the VuePress 2 documentation server with hot-reloading for local development.

```bash
pnpm docs:dev
```

--------------------------------

### Run all linters with pre-commit run -a

Source: https://taskiq-python.github.io/contrib.html

Runs all configured pre-commit linters manually without installing the git hook. Alternatively, install the hook with `pre-commit install`.

```bash
pre-commit run -a
```

--------------------------------

### Install Taskiq via pip

Source: https://taskiq-python.github.io/

Use this command to install the taskiq package using pip.

```bash
pip install taskiq
```

--------------------------------

### Install PostgreSQL result backend

Source: https://taskiq-python.github.io/available-components/result-backends.html

Install the taskiq-postgresql package to use the PostgreSQL result backend.

```bash
pip install taskiq-postgresql
```

--------------------------------

### Install NATS result backend

Source: https://taskiq-python.github.io/available-components/result-backends.html

Install the taskiq-nats package to use the NATS result backend.

```bash
pip install taskiq-nats
```

--------------------------------

### Install aiohttp-deps

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-aiohttp.html

Install the aiohttp-deps library to add FastAPI-like dependency injection to AioHTTP applications.

```bash
pip install "aiohttp-deps"
```

--------------------------------

### Start broker with FastAPI lifespan (recommended)

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-fastapi.html

Recommended way to start and stop the broker using FastAPI's lifespan context. The broker is only started/shutdown in non-worker processes.

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from your_project.taskiq import broker


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    if not broker.is_worker_process:
        await broker.startup()
    yield
    # Shutdown
    if not broker.is_worker_process:
        await broker.shutdown()


app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Async middleware example

Source: https://taskiq-python.github.io/extending-taskiq/middleware.html

Async middleware example. All methods are async and use asyncio.sleep. Implements startup, shutdown, pre_execute, and post_save hooks.

```python
from asyncio import sleep
from typing import Any

from taskiq import TaskiqMessage, TaskiqMiddleware, TaskiqResult


class MyMiddleware(TaskiqMiddleware):
    async def startup(self) -> None:
        print("RUN STARTUP")
        await sleep(1)

    async def shutdown(self) -> None:
        print("RUN SHUTDOWN")
        await sleep(1)

    async def pre_execute(self, message: "TaskiqMessage") -> TaskiqMessage:
        await sleep(1)
        return message

    async def post_save(
        self,
        message: "TaskiqMessage",
        result: "TaskiqResult[Any]",
    ) -> None:
        await sleep(1)
        print("Post save")
```

--------------------------------

### Start taskiq worker

Source: https://taskiq-python.github.io/guide/getting-started.html

Start the taskiq worker process using the broker defined in broker.py.

```bash
taskiq worker broker:broker
```

--------------------------------

### Project structure for task discovery

Source: https://taskiq-python.github.io/guide/architecture-overview.html

Example directory structure showing where task modules might be located.

```text
test_project
├── broker.py
├── submodule
│   └── tasks.py
└── utils
    └── tasks.py
```

--------------------------------

### Start AioPikaBroker

Source: https://taskiq-python.github.io/guide/getting-started.html

AioPikaBroker requires calling startup() before using it. Add this line at the beginning of the main function.

```python
await broker.startup()
```

--------------------------------

### Complete broker.py with AioPikaBroker

Source: https://taskiq-python.github.io/guide/getting-started.html

Complete broker.py example using AioPikaBroker without a result backend. It sends a task, waits for the result, and prints execution time and return value.

```python
# broker.py
import asyncio

from taskiq_aio_pika import AioPikaBroker

broker = AioPikaBroker("amqp://guest:guest@localhost:5672")


@broker.task
async def add_one(value: int) -> int:
    return value + 1


async def main() -> None:
    await broker.startup()
    # Send the task to the broker.
    task = await add_one.kiq(1)
    # Wait for the result.
    result = await task.wait_result(timeout=2)
    print(f"Task execution took: {result.execution_time} seconds.")
    if not result.is_err:
        print(f"Returned value: {result.return_value}")
    else:
        print("Error found while executing task.")
    await broker.shutdown()


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

--------------------------------

### Start broker with FastAPI on_event (deprecated)

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-fastapi.html

Deprecated way to start and stop the broker using FastAPI's on_event handlers. Use the lifespan approach instead.

```python
from fastapi import FastAPI
from your_project.taskiq import broker

app = FastAPI()


@app.on_event("startup")
async def app_startup():
    if not broker.is_worker_process:
        await broker.startup()


@app.on_event("shutdown")
async def app_shutdown():
    if not broker.is_worker_process:
        await broker.shutdown()
```

--------------------------------

### Complete broker.py with Redis result backend

Source: https://taskiq-python.github.io/guide/getting-started.html

Complete broker.py example with RedisAsyncResultBackend. It sends a task, waits for the result, and prints execution time and return value.

```python
# broker.py
import asyncio

from taskiq_aio_pika import AioPikaBroker
from taskiq_redis import RedisAsyncResultBackend

broker = AioPikaBroker(
    "amqp://guest:guest@localhost:5672",
).with_result_backend(RedisAsyncResultBackend("redis://localhost"))


@broker.task
async def add_one(value: int) -> int:
    return value + 1


async def main() -> None:
    await broker.startup()
    # Send the task to the broker.
    task = await add_one.kiq(1)
    # Wait for the result.
    result = await task.wait_result(timeout=2)
    print(f"Task execution took: {result.execution_time} seconds.")
    if not result.is_err:
        print(f"Returned value: {result.return_value}")
    else:
        print("Error found while executing task.")
    await broker.shutdown()


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

--------------------------------

### Install taskiq reload extra with pip

Source: https://taskiq-python.github.io/guide/cli.html

Installs Taskiq with the `reload` extra using pip, enabling hot-reload functionality.

```bash
pip install "taskiq[reload]"
```

--------------------------------

### Run worker with broker path

Source: https://taskiq-python.github.io/guide/architecture-overview.html

Command to start a worker by providing the path to the broker variable. The format is module:variable.

```bash
taskiq worker my_project.broker:mybroker
```

--------------------------------

### RedisScheduleSource usage

Source: https://taskiq-python.github.io/available-components/schedule-sources.html

Shows how to create a RedisScheduleSource and pass it to TaskiqScheduler. Requires installing the taskiq-redis package.

```python
from taskiq_redis import RedisScheduleSource

from taskiq import TaskiqScheduler

redis_source = RedisScheduleSource("redis://localhost:6379/0")
scheduler = TaskiqScheduler(broker, sources=[redis_source])
```

--------------------------------

### Run worker with manual task modules

Source: https://taskiq-python.github.io/guide/architecture-overview.html

Command to start a worker and manually specify all task modules to import.

```bash
taskiq worker test_project.broker:broker test_project.submodule.tasks test_project.utils.tasks
```

--------------------------------

### Use MSGPackSerializer with InMemoryBroker

Source: https://taskiq-python.github.io/guide/message-format.html

Use the MSGPackSerializer with an InMemoryBroker. Requires installing taskiq with the msgpack extra.

```python
from taskiq import InMemoryBroker
from taskiq.serializers import MSGPackSerializer

broker = InMemoryBroker().with_serializer(MSGPackSerializer())
```

--------------------------------

### Class as dependency with default values

Source: https://taskiq-python.github.io/guide/state-and-deps.html

Shows the same class dependency example using default values. The class is inferred from the type hint when using TaskiqDepends() without arguments.

```python
from taskiq import TaskiqDepends


async def db_connection() -> str:
    return "let's pretend as this is a connection"


class MyDAO:
    def __init__(self, db_conn: str = TaskiqDepends(db_connection)) -> None:
        self.db_conn = db_conn

    def get_users(self) -> str:
        return self.db_conn.upper()


def my_task(dao: MyDAO = TaskiqDepends()) -> None:
    print(dao.get_users())

```

--------------------------------

### Build and serve documentation in production mode

Source: https://taskiq-python.github.io/contrib.html

Builds the documentation for production and then serves it locally to preview the production build.

```bash
pnpm docs:build
```

```bash
pnpm docs:serve
```

--------------------------------

### Using kicker to change broker and add labels

Source: https://taskiq-python.github.io/guide/architecture-overview.html

Shows how to use the kicker to send a task with a different broker and additional labels. The task is initially assigned to the first broker, but this example sends it via the second broker with label `delay=1`. The result is then retrieved and printed.

```python
import asyncio

from taskiq.brokers.inmemory_broker import InMemoryBroker

broker = InMemoryBroker()
second_broker = InMemoryBroker()


@broker.task
async def my_async_task() -> None:
    """My lovely task."""
    await asyncio.sleep(1)
    print("Hello")


async def main():
    # This task was initially assigned to broker,
    # but this time it is going to be sent using
    # the second broker with additional label `delay=1`.
    task = await my_async_task.kicker().with_broker(second_broker).with_labels(delay=1).kiq()
    print(await task.get_result())


asyncio.run(main())
```

--------------------------------

### Run Taskiq worker with broker and modules

Source: https://taskiq-python.github.io/guide/cli.html

Starts a Taskiq worker with the specified broker variable and modules containing tasks. The broker is referenced as `mybroker:broker_var`, and the modules are listed as arguments.

```bash
taskiq worker mybroker:broker_var my_project.module1 my_project.module2
```

--------------------------------

### Run worker with file system discovery

Source: https://taskiq-python.github.io/guide/architecture-overview.html

Command to start a worker and let taskiq find all python modules named tasks in the current directory recursively using the -fsd option.

```bash
taskiq worker test_project.broker:broker -fsd
```

--------------------------------

### Store Redis connection pool in state on worker startup

Source: https://taskiq-python.github.io/guide/state-and-deps.html

Example from 'How dependencies are useful'. Stores a Redis connection pool in the global state during worker startup for later use in tasks.

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

```

--------------------------------

### Inject Context into a Task with Annotated or Default Value

Source: https://taskiq-python.github.io/guide/architecture-overview.html

Shows two ways to inject the Context object into a task: using Annotated with TaskiqDepends (Python 3.10+) or using a default value. Both examples create a ZeroMQBroker and define a task that receives the context.

```python
from taskiq import Context, TaskiqDepends, ZeroMQBroker

broker = ZeroMQBroker()


@broker.task
async def my_task(context: Annotated[Context, TaskiqDepends()]):
    ...
```

```python
from taskiq import Context, TaskiqDepends, ZeroMQBroker

broker = ZeroMQBroker()


@broker.task
async def my_task(context: Context = TaskiqDepends()):
    ...
```

--------------------------------

### Register taskiq_cli entry point in setuptools setup.py

Source: https://taskiq-python.github.io/extending-taskiq/cli.html

Register the custom command as a taskiq_cli entry point in a setuptools setup.py file. The entry point name (demo) becomes the subcommand name, and the value points to the MyCommand class.

```python
from setuptools import setup

setup(
    # ...,
    entry_points={
        'taskiq_cli': [
            'demo = my_project.cmd:MyCommand',
        ]
    }
)
```

--------------------------------

### Register taskiq_cli entry point in setuptools pyproject.toml

Source: https://taskiq-python.github.io/extending-taskiq/cli.html

Register the custom command as a taskiq_cli entry point in a setuptools pyproject.toml file. The entry point name (demo) becomes the subcommand name, and the value points to the MyCommand class.

```toml
[project.entry-points.taskiq_cli]
demo = "my_project.cmd:MyCommand"
```

--------------------------------

### Minimal AsyncResultBackend implementation

Source: https://taskiq-python.github.io/extending-taskiq/result-backend.html

Shows the required methods to implement a custom result backend: startup, shutdown, set_result, get_result, and is_result_ready. The get_result method includes a with_logs parameter that is deprecated and will be removed in future releases. It is good practice to skip fetching logs unless with_logs=True is explicitly specified.

```python
from typing import TypeVar

from taskiq import TaskiqResult
from taskiq.abc.result_backend import AsyncResultBackend

_ReturnType = TypeVar("_ReturnType")


class MyResultBackend(AsyncResultBackend[_ReturnType]):
    async def startup(self) -> None:
        """Do something when starting broker."""

    async def shutdown(self) -> None:
        """Do something on shutdown."""

    async def set_result(
        self,
        task_id: str,
        result: TaskiqResult[_ReturnType],
    ) -> None:
        """
        Set result in your backend.

        :param task_id: current task id.
        :param result: result of execution.
        """

    async def get_result(
        self,
        task_id: str,
        with_logs: bool = False,
    ) -> TaskiqResult[_ReturnType]:
        """
        Here you must retrieve result by id.

        Logs is a part of a result.
        Here we have a parameter whether you want to
        fetch result with logs or not, because logs
        can have a lot of info and sometimes it's critical
        to get only needed information.

        :param task_id: id of a task.
        :param with_logs: whether to fetch logs.
        :return: result.
        """
        return ...  # type: ignore

    async def is_result_ready(
        self,
        task_id: str,
    ) -> bool:
        """
        Check if result exists.

        This function must check whether result
        is available in your result backend
        without fetching the result.

        :param task_id: id of a task.
        :return: True if result is ready.
        """
        return ...  # type: ignore

```

--------------------------------

### Skipping first run

Source: https://taskiq-python.github.io/guide/scheduling-tasks.html

Command to start the scheduler with the --skip-first-run flag to avoid immediate execution of tasks that were scheduled before the scheduler starts.

```bash
taskiq scheduler module:scheduler --skip-first-run
```

--------------------------------

### Use CBORSerializer with InMemoryBroker

Source: https://taskiq-python.github.io/guide/message-format.html

Use the CBORSerializer with an InMemoryBroker. Requires installing taskiq with the cbor extra.

```python
from taskiq import InMemoryBroker
from taskiq.serializers import CBORSerializer

broker = InMemoryBroker().with_serializer(CBORSerializer())
```

--------------------------------

### Use ORJSONSerializer with InMemoryBroker

Source: https://taskiq-python.github.io/guide/message-format.html

Use the ORJSONSerializer with an InMemoryBroker. Requires installing taskiq with the orjson extra.

```python
from taskiq import InMemoryBroker
from taskiq.serializers import ORJSONSerializer

broker = InMemoryBroker().with_serializer(ORJSONSerializer())
```

--------------------------------

### Initialize taskiq broker with FastAPI app

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-fastapi.html

Initializes the taskiq broker with the FastAPI application path. Call this in your main broker file. The string path follows the same format as uvicorn (e.g., "my_package.application:app").

```python
from taskiq import ZeroMQBroker
import taskiq_fastapi

broker = ZeroMQBroker()

taskiq_fastapi.init(broker, "my_package.application:app")
```

--------------------------------

### Define task with default value dependency

Source: https://taskiq-python.github.io/guide/testing-taskiq.html

Example of defining a task using default values for dependency injection.

```python
from pathlib import Path
from taskiq import TaskiqDepends

from your_project.tkq import broker


@broker.task
async def modify_path(some_path: Path = TaskiqDepends()):
    return some_path.parent / "taskiq.py"
```

--------------------------------

### Run RabbitMQ with Docker (Windows)

Source: https://taskiq-python.github.io/guide/getting-started.html

Run RabbitMQ in Docker for Windows. Uses caret line continuations instead of backslashes.

```bash
docker run --rm -d ^
    -p "5672:5672" ^
    -p "15672:15672" ^
    --env "RABBITMQ_DEFAULT_USER=guest" ^
    --env "RABBITMQ_DEFAULT_PASS=guest" ^
    --env "RABBITMQ_DEFAULT_VHOST=/" ^
    rabbitmq:3.8.27-management-alpine
```

--------------------------------

### Output of taskiq demo --help

Source: https://taskiq-python.github.io/extending-taskiq/cli.html

Output of running the custom subcommand with --help. It shows the generated usage and the --test option defined in the MyCommand exec method.

```shell
$ taskiq demo --help
usage: demo [-h] [--test TEST]

optional arguments:
  -h, --help   show this help message and exit
  --test TEST  My test parameter.
```

--------------------------------

### Define task with Annotated dependency

Source: https://taskiq-python.github.io/guide/testing-taskiq.html

Example of defining a task using Annotated for dependency injection in Python 3.10+.

```python
from typing import Annotated
from pathlib import Path
from taskiq import TaskiqDepends

from your_project.tkq import broker


@broker.task
async def modify_path(some_path: Annotated[Path, TaskiqDepends()]):
    return some_path.parent / "taskiq.py"
```

--------------------------------

### Register taskiq_cli entry point in Poetry pyproject.toml

Source: https://taskiq-python.github.io/extending-taskiq/cli.html

Register the custom command as a taskiq_cli entry point in a Poetry pyproject.toml file. The entry point name (demo) becomes the subcommand name, and the value points to the MyCommand class.

```toml
[tool.poetry.plugins.taskiq_cli]
demo = "my_project.cmd:MyCommand"
```

--------------------------------

### Disabling dependency cache with default values

Source: https://taskiq-python.github.io/guide/state-and-deps.html

Shows the same use_cache=False example using default values. The result is unpredictable because common_dep is evaluated twice.

```python
import random

from taskiq import TaskiqDepends


def common_dep() -> int:
    return random.randint(1, 10)


def dep1(cd: int = TaskiqDepends(common_dep)) -> int:
    return cd + 1


def dep2(cd: int = TaskiqDepends(common_dep, use_cache=False)) -> int:
    return cd + 2


def my_task(
    d1: int = TaskiqDepends(dep1),
    d2: int = TaskiqDepends(dep2),
) -> int:
    return d1 + d2

```

--------------------------------

### Initialize taskiq-aiogram in tkq.py

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-aiogram.html

Initialize the taskiq-aiogram integration in the broker's main file (tkq.py). The init function takes the broker and paths to the dispatcher and bot instance, and can accept additional bots. The my_task function demonstrates using the bot dependency to send a delayed message.

```python
import asyncio

import taskiq_aiogram
from aiogram import Bot
from taskiq import TaskiqDepends
from taskiq_redis import ListQueueBroker

broker = ListQueueBroker("redis://localhost")

# This line is going to initialize everything.
taskiq_aiogram.init(
    broker,
    # This is path to the dispatcher.
    "bot:dp",
    # This is path to the bot instance.
    "bot:bot",
    # You can specify more bots here.
)


@broker.task(task_name="my_task")
async def my_task(chat_id: int, bot: Bot = TaskiqDepends()) -> None:
    print("I'm a task")
    await asyncio.sleep(4)
    await bot.send_message(chat_id, "task completed")
```

--------------------------------

### Configure AioPikaBroker

Source: https://taskiq-python.github.io/guide/getting-started.html

Replace the InMemoryBroker with AioPikaBroker, connecting to RabbitMQ at the given AMQP URL.

```python
from taskiq_aio_pika import AioPikaBroker

broker = AioPikaBroker('amqp://guest:guest@localhost:5672')
```

--------------------------------

### Custom broker template (AsyncBroker)

Source: https://taskiq-python.github.io/extending-taskiq/broker.html

Template for implementing a custom TaskIQ broker. Override the required `kick` and `listen` methods, and optionally `startup`/`shutdown`. Call `super().__init__()` to set default values. The `listen` method must yield raw bytes or `AckableMessage` objects.

```python
from collections.abc import AsyncGenerator

from taskiq import AckableMessage, AsyncBroker, BrokerMessage


class MyBroker(AsyncBroker):
    def __init__(self) -> None:
        # Please call this super method to set default values to
        # many different fields.
        super().__init__()

    async def startup(self) -> None:
        # Here you can do some startup magic.
        # Like opening a connection.
        return await super().startup()

    async def shutdown(self) -> None:
        # Here you can perform shutdown operations.
        # Like closing connections.
        return await super().shutdown()

    async def kick(self, message: BrokerMessage) -> None:
        # Send a message.message.
        pass

    async def listen(self) -> AsyncGenerator[bytes | AckableMessage, None]:
        while True:
            # Get new message.
            new_message: bytes = ...  # type: ignore
            # Yield it!
            yield new_message
```

--------------------------------

### Get schedule_id from task context labels

Source: https://taskiq-python.github.io/guide/scheduling-tasks.html

Retrieves the schedule_id from the task's context labels. The schedule_id is stored in the message labels and can be accessed inside the task.

```python
@broker.task
async def my_task(context: Context = TaskiqDepends()) -> None:
    schedule_id = context.message.labels.get("schedule_id")
    print("Schedule ID:", schedule_id)
```

--------------------------------

### NATSKeyValueScheduleSource usage

Source: https://taskiq-python.github.io/available-components/schedule-sources.html

Shows how to use NATSKeyValueScheduleSource with TaskiqScheduler. Requires installing the taskiq-nats package. This source does not use the schedule label; use add_schedule method to add schedules.

```python
from taskiq_nats import NATSKeyValueScheduleSource
from taskiq.scheduler import TaskiqScheduler


broker = ...

scheduler = TaskiqScheduler(
    broker=broker,
    sources=[NATSKeyValueScheduleSource(broker)],
)
```

--------------------------------

### Run Redis with Docker (Windows)

Source: https://taskiq-python.github.io/guide/getting-started.html

Run Redis in Docker for Windows. Uses caret line continuations.

```bash
docker run --rm -d ^
    -p "6379:6379" ^
    redis
```

--------------------------------

### Run broker.py with result backend

Source: https://taskiq-python.github.io/guide/getting-started.html

Running broker.py with the Redis result backend prints the correct result (2) and a valid execution time.

```bash
$ python broker.py
Task execution took: 1.0013580322265625e-05 seconds.
Returned value: 2
```

--------------------------------

### Dependency caching with default values

Source: https://taskiq-python.github.io/guide/state-and-deps.html

Shows the same dependency caching example using default values instead of Annotated. The common_dep is evaluated once and shared, so my_task returns 19.

```python
import random

from taskiq import TaskiqDepends


def common_dep() -> int:
    # For example it returns 8
    return random.randint(1, 10)


def dep1(cd: int = TaskiqDepends(common_dep)) -> int:
    # This function will return 9
    return cd + 1


def dep2(cd: int = TaskiqDepends(common_dep)) -> int:
    # This function will return 10
    return cd + 2


def my_task(
    d1: int = TaskiqDepends(dep1),
    d2: int = TaskiqDepends(dep2),
) -> int:
    # This function will return 19
    return d1 + d2

```

--------------------------------

### Test fixture with populate_context

Source: https://taskiq-python.github.io/framework_integrations/taskiq-with-aiohttp.html

Create a pytest fixture that populates the broker's context for InMemoryBroker testing. It starts a TestServer and client, calls taskiq_aiohttp.populate_context, and resets the context after the test.

```python
import taskiq_aiohttp

@pytest.fixture
async def test_client(
    app: web.Application,
) -> AsyncGenerator[TestClient, None]:
    """
    Create a test client.

    This function creates a TestServer
    and a test client for the application.

    Also this fixture populates context
    with needed variables.

    :param app: current application.
    :yield: ready to use client.
    """
    loop = asyncio.get_running_loop()
    server = TestServer(app)
    client = TestClient(server, loop=loop)

    await client.start_server()

    # This is important part.
    # Since InMemoryBroker doesn't
    # run as a worker process, we have to populate
    # broker's context by hand.
    taskiq_aiohttp.populate_context(
        broker=broker,
        server=server.runner.server,
        app=app,
        loop=loop,
    )

    yield client

    broker.custom_dependency_context = {}
    await client.close()
```
Tokens:
Raw