Welcome to RabbitMQ Tools

Asynchronous RabbitMQ Tools for asyncio and Python.

Library Installation

pip install --extra-index-url https://pypi2.abm-jsc.ru rabbitmq-tools-abm

Quick Start

Without Exchanger

import asyncio

from rabbitmq_tools.rabbitmq_connector import RabbitMQConnector
from rabbitmq_tools.rabbitmq_consumer import RabbitMQConsumer


class MyRabbitMQConsumer(RabbitMQConsumer):
    async def _process(self, msg: bytes) -> None:
        print(msg)
        await asyncio.sleep(2)


async def main():
    connector = RabbitMQConnector(
        config=RabbitMQConnector.Config(address="0.0.0.0:5672", username="username", password="password")
    )
    await connector._on_start()
    channel = await connector.get_channel()
    consumer = MyRabbitMQConsumer(
        config=MyRabbitMQConsumer.Config(queue_name="name")
    )
    await consumer.run(channel)


asyncio.run(main())

RabbitMQ Consumer

import asyncio

from aio_pika import ExchangeType

from rabbitmq_tools.rabbitmq_connector import RabbitMQConnector
from rabbitmq_tools.rabbitmq_exchanger import RabbitMQExchanger
from rabbitmq_tools.rabbitmq_consumer import RabbitMQConsumer


class RabbitMQConsumer1(RabbitMQConsumer):
    async def _process(self, msg: bytes) -> None:
        ...


class RabbitMQConsumer2(RabbitMQConsumer):
    async def _process(self, msg: bytes) -> None:
        ...


async def main():
    rabbit_connector = RabbitMQConnector(
        config=RabbitMQConnector.Config(
            address='0.0.0.0:5672',
            username='username',
            password='password'
        )
    )
    await rabbit_connector._on_start()

    rabbitmq_consumer_1 = RabbitMQConsumer1(
        config=RabbitMQConsumer.Config(
            queue_name='queue_name_1'
        )
    )
    rabbitmq_consumer_2 = RabbitMQConsumer2(
        config=RabbitMQConsumer.Config(
            queue_name='queue_name_2'
        )
    )
    rabbit_exchanger = RabbitMQExchanger(
        config=RabbitMQExchanger.Config(
            name='exchange_name_1',
            type=ExchangeType.FANOUT
        ),
        context=RabbitMQExchanger.Context(
            rabbitmq_connector=rabbit_connector
        )
    )
    rabbit_exchanger.register(rabbitmq_consumer_1)
    rabbit_exchanger.register(rabbitmq_consumer_2)

    await rabbit_exchanger._on_start()

    while True:
        await asyncio.sleep(1)
        print('working...')


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