producers, consumers, rabbit_client

src/fake/__init__.py

@@ -1 +0,0 @@
-'''fake data for tests'''

src/messaging/consumers/consumers.py

@@ -0,0 +1,35 @@
+import json
+
+from src.messaging.rabbitmq_client import rabbitmq_client
+
+
+class EmailConsumer:
+    
+    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)
+        self.queue = await self.channel.declare_queue("email_queue", durable=True)
+    
+    async def process_message(self, message) -> None:
+        async with message.process():
+            data = json.loads(message.body)
+            print(f"Отправляю email на {data['email']}")
+    
+    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:
+            async for message in queue_iter:
+                await self.process_message(message)
+
+email_consumer=EmailConsumer()

src/messaging/producers/producers.py

@@ -0,0 +1,33 @@
+import json
+
+import aio_pika
+
+from src.messaging.rabbitmq_client import rabbitmq_client
+
+
+class EmailProducer:
+    
+    def __init__(self) -> None:
+        self.channel = None
+        self.queue = None
+    
+    async def setup(self)->None:
+        
+        self.channel = await rabbitmq_client.get_channel()
+        self.queue = await self.channel.declare_queue("email_queue", durable=True)
+        
+    async def send_welcome_email(self, email:str)->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")
+
+        message=aio_pika.Message(
+            body=json.dumps({"email": email}).encode(),
+            delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
+        )
+        await self.channel.default_exchange.publish(message=message, routing_key=self.queue.name)
+
+email_producer=EmailProducer()

src/messaging/rabbitmq_client.py

@@ -0,0 +1,35 @@
+import aio_pika
+from aio_pika.abc import AbstractChannel, AbstractRobustConnection
+
+from src.models.configs_read.env import env_settings
+
+
+class RabbitMQClient:
+    def __init__(self) -> None:
+        self.connection: AbstractRobustConnection | None = None
+        self.channel: AbstractChannel | None = None
+
+    async def connect(self) -> None:
+        if self.connection is None or self.connection.is_closed:
+            self.connection = await aio_pika.connect_robust(
+                host=env_settings.RABBITMQ_HOST,
+                port=env_settings.RABBITMQ_PORT,
+                login=env_settings.RABBITMQ_LOGIN,
+                password=env_settings.RABBITMQ_PASSWORD,
+            )
+        if self.channel is None or self.channel.is_closed:
+            self.channel = await self.connection.channel()
+
+    async def get_channel(self) -> AbstractChannel:
+        await self.connect()
+        if self.channel is None:
+            raise RuntimeError("Failed to establish RabbitMQ channel")
+        return self.channel
+
+    async def close(self) -> None:
+        if self.channel and not self.channel.is_closed:
+            await self.channel.close()
+        if self.connection and not self.connection.is_closed:
+            await self.connection.close()
+
+rabbitmq_client = RabbitMQClient()

src/models/configs_read/env.py

@@ -20,6 +20,12 @@ class Env(Base):
     REDIS_PORT:int
     REDIS_HOST:str
     
+    RABBITMQ_PASSWORD:str
+    RABBITMQ_LOGIN:str
+    RABBITMQ_HOST:str
+    RABBITMQ_PORT:int
+    RABBITMQ_PORT_UI:int
+    
     PROD_MODE:bool
     
     model_config=SettingsConfigDict(env_file="configs/.env", extra=None)

src/web/protected_routes/auth_routes.py

@@ -74,6 +74,10 @@ async def logout(response:Response,
     response.delete_cookie("refresh_token")
     return await auth.logout(refresh_token, access_token)
 
+from src.messaging.producers.producers import email_producer
+
+ 
 @router.get("")
 async def protected(current_user:UserOut=Depends(require_permissions()))->dict:
+    await email_producer.send_welcome_email("test@test.com")
     return {"protected router": "Hello, this is a protected router"}