topic exchange rabbitmq

src/daemons/base.py

@@ -0,0 +1,9 @@
+from abc import ABC, abstractmethod
+
+
+class BaseDaemon(ABC):
+    name: str
+
+    @abstractmethod
+    async def run(self) -> None:
+        ...

src/daemons/email_daemons.py

@@ -0,0 +1,19 @@
+from src.daemons.base import BaseDaemon
+from src.messaging.consumers.consumers import (
+    reset_email_consumer,
+    welcome_email_consumer,
+)
+
+
+class WelcomeEmailDaemon(BaseDaemon):
+    name = "welcome_email"
+
+    async def run(self) -> None:
+        await welcome_email_consumer.start_consuming()
+
+
+class ResetEmailDaemon(BaseDaemon):
+    name = "reset_email"
+
+    async def run(self) -> None:
+        await reset_email_consumer.start_consuming()

src/daemons/registry.py

@@ -0,0 +1,6 @@
+from src.daemons.email_daemons import ResetEmailDaemon, WelcomeEmailDaemon
+
+DAEMONS = {
+    "welcome_email": WelcomeEmailDaemon,
+    "reset_email": ResetEmailDaemon,
+}

src/messaging/consumers/consumers.py

@@ -1,9 +1,11 @@
 import json
 
+import aio_pika
+
 from src.messaging.rabbitmq_client import rabbitmq_client
 
 
-class EmailConsumer:
+class WelcomeEmailConsumer:
     
     def __init__(self) -> None:
         self.channel = None
@@ -13,23 +15,65 @@ class EmailConsumer:
         
         self.channel = await rabbitmq_client.get_channel()
         await self.channel.set_qos(prefetch_count=10)
-        self.queue = await self.channel.declare_queue("email_queue", durable=True)
+        
+        exchange = await self.channel.declare_exchange("email", aio_pika.ExchangeType.TOPIC)
+        self.queue = await self.channel.declare_queue("queue_welcome_email", durable=True, arguments={"x-queue-type": "quorum"})
+        
+        await self.queue.bind(exchange, routing_key="email.welcome")
     
     async def process_message(self, message) -> None:
+        
         async with message.process():
             data = json.loads(message.body)
-            print(f"Отправляю email на {data['email']}")
+            print(f"Обрабатываю: {data}, метка: {message.routing_key}")
     
     async def start_consuming(self)->None:
         
         if self.queue is None:
             await self.setup()
-        
-        if self.channel is None or self.queue is None:
-            raise RuntimeError("Failed to set up RabbitMQ channel/queue")
-        
-        async with self.queue.iterator() as queue_iter:
+
+        queue = self.queue
+        if queue is None:
+            raise RuntimeError("Failed to set up RabbitMQ queue")
+
+        async with queue.iterator() as queue_iter:
             async for message in queue_iter:
                 await self.process_message(message)
 
-email_consumer=EmailConsumer()
+class ResetEmailConsumer:
+    
+    def __init__(self) -> None:
+        self.channel = None
+        self.queue = None
+    
+    async def setup(self)->None:
+        
+        self.channel = await rabbitmq_client.get_channel()
+        await self.channel.set_qos(prefetch_count=10)
+        
+        exchange = await self.channel.declare_exchange("email", aio_pika.ExchangeType.TOPIC)
+        self.queue = await self.channel.declare_queue("queue_reset_email", durable=True, arguments={"x-queue-type": "quorum"})
+        
+        await self.queue.bind(exchange, routing_key="email.reset")
+    
+    async def process_message(self, message) -> None:
+        
+        async with message.process():
+            data = json.loads(message.body)
+            print(f"Обрабатываю: {data}, метка: {message.routing_key}")
+    
+    async def start_consuming(self)->None:
+        
+        if self.queue is None:
+            await self.setup()
+
+        queue = self.queue
+        if queue is None:
+            raise RuntimeError("Failed to set up RabbitMQ queue")
+
+        async with queue.iterator() as queue_iter:
+            async for message in queue_iter:
+                await self.process_message(message)
+                
+welcome_email_consumer=WelcomeEmailConsumer()
+reset_email_consumer=ResetEmailConsumer()

src/messaging/producers/producers.py

@@ -14,20 +14,28 @@ class EmailProducer:
     async def setup(self)->None:
         
         self.channel = await rabbitmq_client.get_channel()
-        self.queue = await self.channel.declare_queue("email_queue", durable=True)
+        self.exchange = await self.channel.declare_exchange("email", aio_pika.ExchangeType.TOPIC)
         
     async def send_welcome_email(self, email:str)->None:
         
-        if self.queue is None:
-            await self.setup()
+        await self._publish({"email":email}, routing_key="email.welcome")
         
-        if self.channel is None or self.queue is None:
-            raise RuntimeError("Failed to set up RabbitMQ channel/queue")
+    async def send_reset_email(self, email:str)->None:
+        
+        await self._publish({"email":email}, routing_key="email.reset")
+    
+    async def _publish(self,data:dict, routing_key:str)->None:
+        
+        if self.exchange is None:
+            await self.setup()
 
-        message=aio_pika.Message(
-            body=json.dumps({"email": email}).encode(),
+        if self.exchange is None:
+            raise RuntimeError("Failed to set up RabbitMQ exchange")
+
+        message = aio_pika.Message(
+            body=json.dumps(data).encode(),
             delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
         )
-        await self.channel.default_exchange.publish(message=message, routing_key=self.queue.name)
+        await self.exchange.publish(message=message, routing_key=routing_key)
 
 email_producer=EmailProducer()

src/models/configs_read/daemons_json.py

@@ -0,0 +1,12 @@
+from pydantic_settings import SettingsConfigDict
+
+from src.models.configs_read.env import Base
+
+
+class DaemonsConfig(Base):
+    
+    daemons: list[str]
+    
+    model_config = SettingsConfigDict(json_file="configs/daemons.json")
+
+daemons_config = DaemonsConfig() # type: ignore[call-arg]