refactor logging module, add welcome and reset daemons, changed structure of the rabbitmq queues

src/logging/logger.py

@@ -1,8 +1,10 @@
 import asyncio
 import json
 import logging
+from contextvars import ContextVar
 from time import gmtime, perf_counter, strftime
 from typing import cast
+from uuid import uuid4
 
 import aiofiles
 from fastapi import Request
@@ -11,8 +13,28 @@ from starlette.concurrency import iterate_in_threadpool
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.responses import Response, StreamingResponse
 
+request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")
 
-class ProcessingTimeMiddleware(BaseHTTPMiddleware):
+log_queue=asyncio.Queue()
+
+class LogWriter:
+    def __init__(self) -> None:
+        pass
+        
+    async def log_writer(self)->None:
+        while True:
+            log_type, msg=await log_queue.get()
+            await self._write(msg, log_type)
+    
+    async def _write(self, msg:str, type:str)->None:
+        file_time = strftime("%b_%Y", gmtime())
+        current_time = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
+        
+        async with aiofiles.open(f"./logs/{type}_log_{file_time}.txt", "a") as f:
+            await f.write(f"[{current_time}] {msg}\n")
+            
+
+class ProcessingTimeMiddleware(BaseHTTPMiddleware, ):
     async def dispatch(self, request: Request, call_next)->Response:
         start_time = perf_counter()
         response = await call_next(request)
@@ -22,19 +44,26 @@ class ProcessingTimeMiddleware(BaseHTTPMiddleware):
 
 
 class LoggingMiddleware(BaseHTTPMiddleware):
-    async def dispatch(self, request: Request, call_next) -> Response:
-        current_time = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
-        file_time = strftime("%b_%Y", gmtime())
-        client_ip = request.headers.get('x-forwarded-for', '').split(',')[0].strip() or (request.client.host if request.client else 'unknown')
+    
+    async def build_log_line(self, request_id, method, path, status_code, detail, client_ip) -> str:
+        return f"[{request_id}] [{method}] [{path}] [{status_code}] [{detail}] [{client_ip}]"
 
+    async def dispatch(self, request: Request, call_next) -> Response:
+        request_id=str(uuid4())
+        request_id_ctx.set(request_id)
+        client_ip = request.headers.get('x-forwarded-for', '').split(',')[0].strip() or (request.client.host if request.client else 'unknown')
+        method = request.method
+        path=request.url.path
         try:
             response = await call_next(request)
         except Exception as exc:  # noqa: BLE001
-            body = str(exc)
-            async with aiofiles.open(f"./logs/endpoints_log_{file_time}.txt", "a") as file:
-                await file.write(f"[{current_time}] [500] [{body}] [{client_ip}]\n")
-            return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})
-
+            line=await self.build_log_line(request_id=request_id, method=method, path=path, status_code=500, detail=repr(exc), client_ip=client_ip)
+            log_queue.put_nowait(("endpoints", line))
+            return JSONResponse(
+                status_code=500,
+                content={"detail": "Internal Server Error", "request_id": request_id}
+            )
+            
         streaming_response = cast(StreamingResponse, response)
 
         chunks = []
@@ -49,23 +78,16 @@ class LoggingMiddleware(BaseHTTPMiddleware):
             body = parsed.get("detail", None) if not isinstance(parsed, bool) else None
         except (json.JSONDecodeError, TypeError):
             body = None
-
-        async with aiofiles.open(f"./logs/endpoints_log_{file_time}.txt", "a") as file:
-            await file.write(f"[{current_time}] [{response.status_code}] [{body}] [{client_ip}]\n")
-
+            
+        line=await self.build_log_line(request_id=request_id, method=method, path=path, status_code=response.status_code, detail=body, client_ip=client_ip)
+        log_queue.put_nowait(("endpoints", line))
+    
         return response
 
 
-class LoggerDB(logging.Handler):
+class LoggerDB(logging.Handler, LogWriter):
     
     def emit(self, record: logging.LogRecord) -> None:
         msg = self.format(record)
-        asyncio.create_task(self._write(msg))
-
-    async def _write(self, msg: str) -> None:
-        
-        file_time = strftime("%b_%Y", gmtime())
-        current_time = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
-        
-        async with aiofiles.open(f"./logs/sql_log_{file_time}.txt", "a") as f:
-            await f.write(f"[{current_time}] {msg}\n")
+        rid=request_id_ctx.get()
+        log_queue.put_nowait(("sql",f"[{rid}] {msg}"))

src/messaging/consumers/consumers.py

@@ -1,8 +1,8 @@
 import json
-
-import aio_pika
+import smtplib
 
 from src.messaging.rabbitmq_client import rabbitmq_client
+from src.service.email.email_welcome import DaemonEmailSender
 
 
 class WelcomeEmailConsumer:
@@ -10,22 +10,33 @@ class WelcomeEmailConsumer:
     def __init__(self) -> None:
         self.channel = None
         self.queue = None
-    
-    async def setup(self)->None:
+        self.daemon=DaemonEmailSender()
         
+    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_welcome_email", durable=True, arguments={"x-queue-type": "quorum"})
-        
-        await self.queue.bind(exchange, routing_key="email.welcome")
+        self.queue = await self.channel.get_queue("queue_welcome_email")
+
     
     async def process_message(self, message) -> None:
         
-        async with message.process():
+        async with message.process(ignore_processed=True):
             data = json.loads(message.body)
             print(f"Обрабатываю: {data}, метка: {message.routing_key}")
+            try:
+                await self.daemon.send_email(data.get("email"))
+            except (
+                smtplib.SMTPServerDisconnected,
+                smtplib.SMTPConnectError,
+                TimeoutError,
+                ConnectionRefusedError,
+            ) as exc:
+                print(f"transient error, retrying: {exc!r}")
+                await message.nack(requeue=True)
+            except Exception as exc:  # noqa: BLE001
+                print(f"permanent error, sending to DLQ: {exc!r}")
+                await message.nack(requeue=False)
+
     
     async def start_consuming(self)->None:
         
@@ -46,15 +57,12 @@ class ResetEmailConsumer:
         self.channel = None
         self.queue = None
     
-    async def setup(self)->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")
+        self.queue = await self.channel.get_queue("queue_reset_email")
+
     
     async def process_message(self, message) -> None:
         

src/messaging/producers/producers.py

@@ -11,16 +11,10 @@ class EmailProducer:
         self.channel = None
         self.exchange = None
     
-    async def setup(self)->None:
-        
+    async def setup(self) -> None:
         self.channel = await rabbitmq_client.get_channel()
-        self.exchange = await self.channel.declare_exchange("email", aio_pika.ExchangeType.TOPIC)
-        
-        welcome_queue = await self.channel.declare_queue("queue_welcome_email", durable=True, arguments={"x-queue-type": "quorum"})
-        await welcome_queue.bind(self.exchange, routing_key="email.welcome")
+        self.exchange = await self.channel.get_exchange("email")
 
-        reset_queue = await self.channel.declare_queue("queue_reset_email", durable=True, arguments={"x-queue-type": "quorum"})
-        await reset_queue.bind(self.exchange, routing_key="email.reset")
         
     async def send_welcome_email(self, email:str)->None:
         

src/messaging/topology_setup.py

@@ -0,0 +1,22 @@
+from src.models.rabbitmq_models.email import Topology
+
+
+async def apply_topology(channel, topology: Topology) -> None:
+    dlx = await channel.declare_exchange("email.dlx", "direct")
+
+    for exch_spec in topology.exchanges:
+        exchange = await channel.declare_exchange(exch_spec.name, exch_spec.type)
+        for q in exch_spec.queues:
+            queue = await channel.declare_queue(
+                q.name,
+                durable=True,
+                arguments={
+                    "x-queue-type": "quorum",
+                    "x-dead-letter-exchange": "email.dlx",
+                },
+            )
+            await queue.bind(exchange, routing_key=q.routing_key)
+
+            if q.dlx_queue:
+                dlq = await channel.declare_queue(q.dlx_queue, durable=True)
+                await dlq.bind(dlx, routing_key=q.routing_key)

src/models/configs_read/env.py

@@ -26,6 +26,11 @@ class Env(Base):
     RABBITMQ_PORT:int
     RABBITMQ_PORT_UI:int
     
+    EMAIL_PORT:int
+    SMTP_SERVER:str
+    EMAIL_LOGIN:str
+    EMAIL_PASSWORD:str
+    
     PROD_MODE:bool
     
     model_config=SettingsConfigDict(env_file="configs/.env", extra=None)

src/models/rabbitmq_models/email.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel
+
+
+class QueueSpec(BaseModel):
+    name:str
+    routing_key:str
+    dlx_queue:str|None = None
+
+class ExchangeSpec(BaseModel):
+    name:str
+    type:str = "topic"
+    queues: list[QueueSpec]
+
+class Topology(BaseModel):
+    exchanges:list[ExchangeSpec]
+    
+email_topology=Topology(exchanges=[
+    ExchangeSpec(name="email", queues=[
+        QueueSpec(name="queue_welcome_email", routing_key="email.welcome", dlx_queue="queue_welcome_email.dlq"),
+        QueueSpec(name="queue_reset_email", routing_key="email.reset", dlx_queue="queue_reset_email.dlq"),
+    ]),
+])
+    
+    

src/service/email/email_welcome.py

@@ -0,0 +1,40 @@
+import smtplib
+import ssl
+from email.message import EmailMessage
+
+from src.models.configs_read.env import env_settings
+from src.service.email.jinja_env import jinja_env
+
+
+class DaemonEmailSender:
+    def __init__(self) -> None:
+        pass
+    
+    async def send_email(self, target_email:str)->None:
+        context = ssl.create_default_context()
+        with smtplib.SMTP_SSL(
+            env_settings.SMTP_SERVER, env_settings.EMAIL_PORT, context=context
+        ) as server:
+            server.login(env_settings.EMAIL_LOGIN, env_settings.EMAIL_PASSWORD)
+            msg=await self.build_email_message(target_email)
+            server.send_message(msg)
+            
+    async def build_email_message(self, target_email: str) -> EmailMessage:
+        template = jinja_env.get_template("welcome.html")
+        html_body = template.render(name=target_email)
+
+        text_body = (
+            f"Добро пожаловать, {target_email}!\n\n"
+            "Спасибо за регистрацию на проекте «The DisExcel». "
+            "Мы рады, что вы с нами — учётная запись уже готова к работе.\n\n"
+            "Если вы не регистрировались на The DisExcel, просто проигнорируйте это письмо."
+        )
+
+        msg = EmailMessage()
+        msg["to"] = target_email
+        msg["from"] = env_settings.EMAIL_LOGIN
+        msg["subject"] = "Добро пожаловать"
+        msg.set_content(text_body)
+        msg.add_alternative(html_body, subtype="html")
+        return msg
+

src/service/email/jinja_env.py

@@ -0,0 +1,6 @@
+from jinja2 import Environment, FileSystemLoader
+
+jinja_env = Environment(
+    loader=FileSystemLoader("src/service/email/templates"),
+    autoescape=True,
+)

src/service/email/templates/reset.html

@@ -0,0 +1,55 @@
+<!doctype html>
+<html lang="ru">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>Пароль сброшен</title>
+</head>
+<body style="margin:0; padding:0; background-color:#f4f4f7; font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;">
+  <table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background-color:#f4f4f7; padding:40px 0;">
+    <tr>
+      <td align="center">
+        <table role="presentation" width="480" cellpadding="0" cellspacing="0" style="background-color:#ffffff; border-radius:12px; overflow:hidden; box-shadow:0 2px 8px rgba(0,0,0,0.06);">
+
+          <tr>
+            <td style="background-color:#dc2626; padding:32px 40px; text-align:center;">
+              <h1 style="margin:0; color:#ffffff; font-size:22px; font-weight:600;">The DisExcel</h1>
+            </td>
+          </tr>
+
+          <tr>
+            <td style="padding:40px;">
+              <h2 style="margin:0 0 16px 0; color:#111827; font-size:20px;">Пароль был сброшен</h2>
+              <p style="margin:0 0 24px 0; color:#4b5563; font-size:15px; line-height:1.6;">
+                Пароль от вашей учётной записи в «The DisExcel» был сброшен. Ниже — временный пароль для входа. Рекомендуем сменить его на свой сразу после входа в аккаунт.
+              </p>
+
+              <table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background-color:#f9fafb; border:1px solid #e5e7eb; border-radius:8px;">
+                <tr>
+                  <td style="padding:20px; text-align:center;">
+                    <span style="display:block; margin:0 0 4px 0; color:#9ca3af; font-size:12px; text-transform:uppercase; letter-spacing:0.05em;">Новый пароль</span>
+                    <span style="display:inline-block; font-family: 'Courier New', monospace; font-size:20px; font-weight:700; color:#111827; letter-spacing:0.05em;">Тест1234!</span>
+                  </td>
+                </tr>
+              </table>
+
+              <p style="margin:24px 0 0 0; color:#9ca3af; font-size:13px; line-height:1.5;">
+                Если вы не запрашивали сброс пароля, срочно свяжитесь с поддержкой — возможно, кто-то получил доступ к вашей учётной записи.
+              </p>
+            </td>
+          </tr>
+
+          <tr>
+            <td style="padding:24px 40px; background-color:#f9fafb; border-top:1px solid #e5e7eb;">
+              <p style="margin:0; color:#9ca3af; font-size:12px; line-height:1.5;">
+                Это автоматическое уведомление от The DisExcel. Не пересылайте это письмо никому.
+              </p>
+            </td>
+          </tr>
+
+        </table>
+      </td>
+    </tr>
+  </table>
+</body>
+</html>

src/service/email/templates/welcome.html

@@ -0,0 +1,52 @@
+<!doctype html>
+<html lang="ru">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>Добро пожаловать</title>
+</head>
+<body style="margin:0; padding:0; background-color:#f4f4f7; font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;">
+  <table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background-color:#f4f4f7; padding:40px 0;">
+    <tr>
+      <td align="center">
+        <table role="presentation" width="480" cellpadding="0" cellspacing="0" style="background-color:#ffffff; border-radius:12px; overflow:hidden; box-shadow:0 2px 8px rgba(0,0,0,0.06);">
+
+          <tr>
+            <td style="background-color:#4f46e5; padding:32px 40px; text-align:center;">
+              <h1 style="margin:0; color:#ffffff; font-size:22px; font-weight:600;">The DisExcel</h1>
+            </td>
+          </tr>
+
+          <tr>
+            <td style="padding:40px;">
+              <h2 style="margin:0 0 16px 0; color:#111827; font-size:20px;">Добро пожаловать, {{ name }}</h2>
+              <p style="margin:0 0 24px 0; color:#4b5563; font-size:15px; line-height:1.6;">
+                Спасибо за регистрацию на проекте «The DisExcel». Мы рады, что вы с нами — учётная запись уже готова к работе.
+              </p>
+
+              <table role="presentation" cellpadding="0" cellspacing="0" align="center" style="margin:0 auto;">
+                <tr>
+                  <td style="border-radius:8px; background-color:#4f46e5;">
+                    <a href="#" style="display:inline-block; padding:12px 28px; color:#ffffff; font-size:14px; font-weight:600; text-decoration:none;">
+                      Перейти в аккаунт
+                    </a>
+                  </td>
+                </tr>
+              </table>
+            </td>
+          </tr>
+
+          <tr>
+            <td style="padding:24px 40px; background-color:#f9fafb; border-top:1px solid #e5e7eb;">
+              <p style="margin:0; color:#9ca3af; font-size:12px; line-height:1.5;">
+                Если вы не регистрировались на The DisExcel, просто проигнорируйте это письмо.
+              </p>
+            </td>
+          </tr>
+
+        </table>
+      </td>
+    </tr>
+  </table>
+</body>
+</html>

src/web/protected_routes/auth_routes.py

@@ -79,5 +79,5 @@ 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")
+    await email_producer.send_welcome_email(current_user.email)
     return {"protected router": "Hello, this is a protected router"}