Metadata-Version: 2.4
Name: taskiq-aio-pika
Version: 0.5.0
Summary: RabbitMQ broker for taskiq
Keywords: taskiq,tasks,distributed,async,aio-pika
Author: Pavel Kirilin
Author-email: Pavel Kirilin <win10@list.ru>
License-Expression: MIT
License-File: LICENSE
Classifier: Typing :: Typed
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Operating System :: OS Independent
Classifier: Intended Audience :: Developers
Classifier: Topic :: System :: Networking
Classifier: Development Status :: 3 - Alpha
Requires-Dist: taskiq>=0.11.20,<1
Requires-Dist: aio-pika>=9.0.0
Maintainer: Pavel Kirilin
Maintainer-email: Pavel Kirilin <win10@list.ru>
Requires-Python: >=3.10, <4
Description-Content-Type: text/markdown

# AioPika broker for taskiq

[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/taskiq-aio-pika?style=for-the-badge)](https://pypi.org/project/taskiq-aio-pika/)
[![PyPI](https://img.shields.io/pypi/v/taskiq-aio-pika?style=for-the-badge)](https://pypi.org/project/taskiq-aio-pika/)
[![PyPI - Downloads](https://img.shields.io/pypi/dm/taskiq-aio-pika?style=for-the-badge)](https://pypistats.org/packages/taskiq-aio-pika)

This library provides you with aio-pika broker for taskiq.

Features:
- Supports delayed messages using dead-letter queues or RabbitMQ delayed message exchange plugin.
- Supports message priorities.

Usage example:

```python
from taskiq_aio_pika import AioPikaBroker

broker = AioPikaBroker()

@broker.task
async def test() -> None:
    print("nothing")

```

## Delays

### Default delays

To send delayed message, you have to specify delay label. You can do it with `task` decorator, or by using kicker.

In this type of delay we are using additional queue with `expiration` parameter. After declared time message will be deleted from `delay` queue and sent to the main queue. For example:

```python
broker = AioPikaBroker()

@broker.task(delay=3)
async def delayed_task() -> int:
    return 1

async def main():
    await broker.startup()
    # This message will be received by workers
    # After 3 seconds delay.
    await delayed_task.kiq()

    # This message is going to be received after the delay in 4 seconds.
    # Since we overridden the `delay` label using kicker.
    await delayed_task.kicker().with_labels(delay=4).kiq()

    # This message is going to be send immediately. Since we deleted the label.
    await delayed_task.kicker().with_labels(delay=None).kiq()

    # Of course the delay is managed by rabbitmq, so you don't
    # have to wait delay period before message is going to be sent.
```

### Delays with `rabbitmq-delayed-message-exchange` plugin

First of all please make sure that your RabbitMQ server has [rabbitmq-delayed-message-exchange plugin](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange) installed.

Also you need to configure you broker by passing `delayed_message_exchange_plugin=True` to broker.

This plugin can handle tasks with different delay times well, and the delay based on dead letter queue is suitable for tasks with the same delay time. For example:

```python
broker = AioPikaBroker(
    delayed_message_exchange_plugin=True,
)

@broker.task(delay=3)
async def delayed_task() -> int:
    return 1

async def main():
    await broker.startup()
    # This message will be received by workers
    # After 3 seconds delay.
    await delayed_task.kiq()

    # This message is going to be received after the delay in 4 seconds.
    # Since we overridden the `delay` label using kicker.
    await delayed_task.kicker().with_labels(delay=4).kiq()
```

## Priorities

You can define priorities for messages using `priority` label. Messages with higher priorities are delivered faster.
But to use priorities you need to define `max_priority` of the main queue, by passing `max_priority` parameter in broker's init. This parameter sets maximum priority for the queue and declares it as the priority queue.

Before doing so please read the [documentation](https://www.rabbitmq.com/priority.html#behaviour) about what
downsides you get by using prioritized queues.

```python
broker = AioPikaBroker(max_priority=10)

# We can define default priority for tasks.
@broker.task(priority=2)
async def prio_task() -> int:
    return 1

async def main():
    await broker.startup()
    # This message has priority = 2.
    await prio_task.kiq()

    # This message is going to have priority 4.
    await prio_task.kicker().with_labels(priority=4).kiq()

    # This message is going to have priority 0.
    await prio_task.kicker().with_labels(priority=None).kiq()
```

## Configuration

AioPikaBroker parameters:

* `url` - url to rabbitmq. If None, "amqp://guest:guest@localhost:5672" is used.
* `result_backend` - custom result backend.
* `task_id_generator` - custom task_id genertaor.
* `exchange_name` - name of exchange that used to send messages.
* `exchange_type` - type of the exchange. Used only if `declare_exchange` is True.
* `queue_name` - queue that used to get incoming messages.
* `routing_key` - that used to bind that queue to the exchange.
* `declare_exchange` - whether you want to declare new exchange if it doesn't exist.
* `max_priority` - maximum priority for messages.
* `delay_queue_name` - custom delay queue name. This queue is used to deliver messages with delays.
* `dead_letter_queue_name` - custom dead letter queue name.
    This queue is used to receive negatively acknowledged messages from the main queue.
* `qos` - number of messages that worker can prefetch.
* `declare_queues` - whether you want to declare queues even on client side. May be useful for message persistence.
* `declare_queues_kwargs` - see [Custom Queue Arguments](#custom-queue-arguments) for more details.

## Custom Queue Arguments

You can pass custom arguments to the underlying RabbitMQ queue declaration by using the `declare_queues_kwargs` parameter of `AioPikaBroker`. If you want to set specific queue arguments (such as RabbitMQ extensions or custom behaviors), provide them in the `arguments` dictionary inside `declare_queues_kwargs`.

These arguments will be merged with the default arguments used by the broker
(such as dead-lettering and priority settings). If there are any conflicts, the values you provide will take precedence over the broker's defaults. Example:

```python
broker = AioPikaBroker(
    declare_queues_kwargs={
        "arguments": {
            "x-message-ttl": 60000,  # Set message TTL to 60 seconds
            "x-queue-type": "quorum",  # Use quorum queue type
        }
    }
)
```

This will ensure that the queue is created with your custom arguments, in addition to the broker's defaults.
