Inject AMQP Publisher into AMQP MessageBroker (#20)

* Inject AMQP Publisher into AMQP MessageBroker

* Update version

* Provide adapters protocol

---------

Co-authored-by: m.bahov <m.bahov@vkteam.ru>

Author: 1owkeyme

pyproject.toml

@@ -29,7 +29,7 @@ maintainers = [{name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru"}]
 name = "python-cqrs"
 readme = "README.md"
 requires-python = ">=3.10"
-version = "3.1.1"
+version = "4.0.0"
 
 [project.optional-dependencies]
 dev = [

src/cqrs/adapters/amqp.py

@@ -2,6 +2,8 @@
 import typing
 from functools import partial
 
+from cqrs.adapters import protocol
+
 import aio_pika
 from aio_pika import abc, pool
 
@@ -10,46 +12,28 @@ async def connection_pool_factory(url: str) -> abc.AbstractRobustConnection:
     return await aio_pika.connect_robust(url=url)
 
 
-async def channel_pool_factory(connection_pool: pool.Pool) -> aio_pika.Channel:
+async def channel_pool_factory(
+    connection_pool: pool.Pool[aio_pika.abc.AbstractConnection],
+) -> aio_pika.abc.AbstractChannel:
     async with connection_pool.acquire() as connection:
         return await connection.channel()
 
 
-class AMQPPublisher:
-    def __init__(self, url: str, max_connection_pool_size=2, max_channel_pool_size=10):
-        self.url = url
-        self.max_connection_pool_size = max_connection_pool_size
-        self.max_channel_pool_size = max_channel_pool_size
-        self.connection_pool: pool.Pool = pool.Pool(
-            partial(connection_pool_factory, url=url),
-            max_size=self.max_connection_pool_size,
-        )
-        self.channel_pool: pool.Pool = pool.Pool(
-            partial(channel_pool_factory, connection_pool=self.connection_pool),
-            max_size=self.max_channel_pool_size,
-        )
+class AMQPPublisher(protocol.AMQPPublisher):
+    def __init__(self, channel_pool: pool.Pool[aio_pika.abc.AbstractChannel]):
+        self.channel_pool = channel_pool
 
     async def publish(self, message: abc.AbstractMessage, queue_name: str, exchange_name: str) -> None:
         async with self.channel_pool.acquire() as channel:
-            queue: aio_pika.Queue = await channel.declare_queue(queue_name)
-            exchange: aio_pika.Exchange = await channel.declare_exchange(exchange_name, type="direct", auto_delete=True)
+            queue = await channel.declare_queue(queue_name)
+            exchange = await channel.declare_exchange(exchange_name, type="direct", auto_delete=True)
             await queue.bind(exchange=exchange, routing_key=queue_name)
             await exchange.publish(message=message, routing_key=queue_name)
 
 
-class AMQPConsumer:
-    def __init__(self, url: str, max_connection_pool_size=2, max_channel_pool_size=10):
-        self.url = url
-        self.max_connection_pool_size = max_connection_pool_size
-        self.max_channel_pool_size = max_channel_pool_size
-        self.connection_pool: pool.Pool = pool.Pool(
-            partial(connection_pool_factory, url=url),
-            max_size=self.max_connection_pool_size,
-        )
-        self.channel_pool: pool.Pool = pool.Pool(
-            partial(channel_pool_factory, connection_pool=self.connection_pool),
-            max_size=self.max_channel_pool_size,
-        )
+class AMQPConsumer(protocol.AMQPConsumer):
+    def __init__(self, channel_pool: pool.Pool[aio_pika.abc.AbstractChannel]):
+        self.channel_pool = channel_pool
 
     async def consume(
         self,
@@ -59,5 +43,41 @@ async def consume(
         async with self.channel_pool.acquire() as channel:
             await channel.set_qos(prefetch_count=1)
             queue = await channel.declare_queue(queue_name)
-            await queue._consume(handler)
+            await queue.consume(handler)
             await asyncio.Future()
+
+
+def amqp_publisher_factory(
+    url: typing.Text,
+    max_connection_pool_size: int = 2,
+    max_channel_pool_size: int = 10,
+) -> AMQPPublisher:
+    max_connection_pool_size = max_connection_pool_size
+    max_channel_pool_size = max_channel_pool_size
+    connection_pool = pool.Pool(
+        partial(connection_pool_factory, url=url),
+        max_size=max_connection_pool_size,
+    )
+    channel_pool = pool.Pool(
+        partial(channel_pool_factory, connection_pool=connection_pool),
+        max_size=max_channel_pool_size,
+    )
+    return AMQPPublisher(channel_pool=channel_pool)
+
+
+def amqp_consumer_factory(
+    url: typing.Text,
+    max_connection_pool_size: int = 2,
+    max_channel_pool_size: int = 10,
+) -> AMQPConsumer:
+    max_connection_pool_size = max_connection_pool_size
+    max_channel_pool_size = max_channel_pool_size
+    connection_pool = pool.Pool(
+        partial(connection_pool_factory, url=url),
+        max_size=max_connection_pool_size,
+    )
+    channel_pool = pool.Pool(
+        partial(channel_pool_factory, connection_pool=connection_pool),
+        max_size=max_channel_pool_size,
+    )
+    return AMQPConsumer(channel_pool=channel_pool)

src/cqrs/adapters/kafka.py

@@ -4,11 +4,13 @@
 import ssl
 import typing
 
+from cqrs.adapters import protocol
+from cqrs.serializers import default
+
 import aiokafka
 import retry_async
 from aiokafka import errors
 
-from cqrs.serializers import default
 
 __all__ = (
     "KafkaProducer",
@@ -45,16 +47,7 @@
 Serializer = typing.Callable[[typing.Any], typing.ByteString | None]
 
 
-class _Singleton(type):
-    _instances = {}
-
-    def __call__(cls, *args, **kwargs):
-        if cls not in cls._instances:
-            cls._instances[cls] = super(_Singleton, cls).__call__(*args, **kwargs)
-        return cls._instances[cls]
-
-
-class KafkaProducer(metaclass=_Singleton):
+class KafkaProducer(protocol.KafkaProducer):
     def __init__(
         self,
         producer: aiokafka.AIOKafkaProducer,

src/cqrs/adapters/protocol.py

@@ -0,0 +1,28 @@
+import typing
+
+import aio_pika
+
+
+class KafkaProducer(typing.Protocol):
+    async def produce(
+        self,
+        topic: typing.Text,
+        message: typing.Any,
+    ) -> None: ...
+
+
+class AMQPPublisher(typing.Protocol):
+    async def publish(
+        self,
+        message: aio_pika.abc.AbstractMessage,
+        queue_name: str,
+        exchange_name: str,
+    ) -> None: ...
+
+
+class AMQPConsumer(typing.Protocol):
+    async def consume(
+        self,
+        handler: typing.Callable[[aio_pika.abc.AbstractIncomingMessage], typing.Awaitable[None]],
+        queue_name: str,
+    ) -> None: ...

src/cqrs/message_brokers/amqp.py

@@ -3,13 +3,13 @@
 import aio_pika
 import orjson
 
-from cqrs.adapters import amqp
+from cqrs.adapters import protocol as adapters_protocol
 from cqrs.message_brokers import protocol
 
 
 class AMQPMessageBroker(protocol.MessageBroker):
-    def __init__(self, dsn: str, exchange_name: str, pika_log_level: str = "ERROR"):
-        self.publisher = amqp.AMQPPublisher(url=dsn)
+    def __init__(self, publisher: adapters_protocol.AMQPPublisher, exchange_name: str, pika_log_level: str = "ERROR"):
+        self.publisher = publisher
         self.exchange_name = exchange_name
         logging.getLogger("aiormq").setLevel(pika_log_level)
         logging.getLogger("aio_pika").setLevel(pika_log_level)

src/cqrs/message_brokers/kafka.py

@@ -1,14 +1,14 @@
 import logging
 import typing
 
-from cqrs.adapters import kafka
+from cqrs.adapters import protocol as adapters_protocol
 from cqrs.message_brokers import protocol
 
 
 class KafkaMessageBroker(protocol.MessageBroker):
     def __init__(
         self,
-        producer: kafka.KafkaProducer,
+        producer: adapters_protocol.KafkaProducer,
         aiokafka_log_level: typing.Text = "ERROR",
     ):
         self._producer = producer

src/cqrs/producer.py

@@ -37,7 +37,7 @@ async def send_message(self, event: repository_protocol.OutboxedEvent):
                     message_name=event.event.event_name,
                     message_id=event.event.event_id,
                     topic=event.topic,
-                    payload=event.event,
+                    payload=event.event.model_dump(),
                 ),
             )
         except Exception as error: