aiocarrot

aiocarrot is a fully asynchronous framework for working with the RabbitMQ message broker

Source Code: https://github.com/d3nbr0/aiocarrot

The key features are:

• Completely asynchronous - aiocarrot has the aiopika library running under the hood
• Fast to code - the framework allows you to reduce the amount of code in your project, as well as speed up its development
• Fields validation - aiocarrot supports field validation using pydantic
• Scheduler - describe the task launch period in the cron format

Requirements

The following dependencies are required for aiocarrot to work:

• aio-pika for working with RabbitMQ
• pydantic for fields validation
• ujson for sending and receiving messages
• loguru for logging :)

Installation

Create and activate virtual environment and then install aiocarrot:

pip install aiocarrot

Example

Create a file main.py with:

from aiocarrot import Carrot, Consumer

import asyncio


consumer = Consumer()


@consumer.message(name='multiply')
async def multiply(first_number: int, second_number: int) -> None:
    print('Result is:', first_number * second_number)


async def main() -> None:
    carrot = Carrot(url='amqp://guest:guest@127.0.0.1/', queue_name='sample')
    carrot.setup_consumer(consumer)
    await carrot.run()


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

Then run it with:

python main.py

Now you have created a consumer with the ability to receive a "multiply" task

Produce message

If you want to send a message, use this:

from aiocarrot import Carrot

import asyncio


async def main() -> None:
    carrot = Carrot(url='amqp://guest:guest@127.0.0.1:5672/', queue_name='sample')
    
    await carrot.send('multiply', first_number=10, second_number=20)


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

Scheduler

You can use the scheduler to run your tasks automatically.
You can schedule a task using the cron format. For example, to run a task once every 15 minutes: */15 * * * *

@consumer.message(name='example.scheduler', schedule='*/15 * * * *')
async def scheduler_message(value: Optional[int] = None) -> None:
    print('Your value is:', value or 0)

NOTE: The scheduler does not support working with arguments that do not have default values. If you want to schedule a message and it has arguments, make sure they all have a default value.

You can find more examples here

It's very simple to use. Enjoy!