add logging and retry to connect to the rabbitmq server

CLAUDE.md

@@ -0,0 +1,112 @@
+# The DisExcel Project — Context for Claude Code
+
+Backend: FastAPI + SQLAlchemy (async) + Pydantic v2 + PostgreSQL. JWT auth
+(access + refresh tokens), RBAC permissions, Redis cache, RabbitMQ background
+workers. Python >=3.13,<4.0, Poetry for dependency management.
+
+## Auth & Permissions
+
+- Users have `direct_permissions` (list of `Permissions`) and `group`
+  (list of `PermissionsGroups`, each with its own `permissions`) —
+  many-to-many both ways.
+- Effective permissions = `direct_permissions ∪ (union of all groups' permissions)`.
+- `require_permissions(*permissions)` in `src/web/protected_routes` is a
+  FastAPI dependency factory — wraps `CurrentUserService.get_current_user`.
+  Call with no args (`require_permissions()`) for "just authenticated, no
+  specific permission needed".
+- Access tokens carry a `jti` claim. Logout writes `revoked_access_token:{jti}`
+  to Redis with TTL = remaining token lifetime — `get_current_user` checks
+  this key before anything else.
+- `secure` flag on refresh_token cookie is driven by `env_settings.PROD_MODE`
+  (bool) — `False` locally/tests so cookies work over plain HTTP, `True` in
+  prod.
+
+## Redis (`src/cache/`)
+
+- `RedisClient(redis.Redis)` — module-level shared singleton, subclasses
+  `redis.Redis` directly (inherits all commands, no manual wrapping needed).
+- Three uses: permissions is-cache was considered and rejected (no real DB
+  savings — `get_user_by_id` already eager-loads everything via `selectin`
+  in one call); rate limiting on login (`RateLimit.rate_limit(ip)` —
+  `INCR` + `EXPIRE` on first attempt, blocks >5/60s); access-token revoke
+  blacklist (see above).
+- Rate limit is only triggered inside `except HTTPException` on `/protected/token`
+  — i.e. only on failed logins, not successful ones (otherwise legitimate
+  repeated logins would trip it).
+
+## RabbitMQ (`src/messaging/`)
+
+- `RabbitMQClient` — shared class, lazy `connect()` (can't be async `__init__`),
+  holds one `connection` + one `channel`, `get_channel()` ensures setup.
+- Topic exchange named `"email"`. Each message type gets its own routing key
+  (`email.welcome`, `email.reset`) and its own durable queue
+  (`queue_welcome_email`, `queue_reset_email`), each queue explicitly bound
+  to the exchange with its own key.
+- **Important**: both producer and consumer must declare/bind the queues —
+  if only the consumer does it and the consumer has never run, `publish`
+  on a not-yet-existing queue silently loses the message. `EmailProducer.setup()`
+  also declares+binds both queues defensively.
+- `message.process()` async context manager = manual ack (auto-ack on
+  success, requeue/nack on exception) — this is the right choice for email,
+  not `no_ack=True`.
+- Email templates: Jinja2, inline CSS (email clients don't support `<style>`
+  reliably), `EmailMessage` with `set_content()` (plain-text fallback) +
+  `add_alternative(html, subtype="html")`.
+
+## Daemons / workers (`src/daemons/`)
+
+- `BaseDaemon` ABC (`name` + async `run()`), one subclass per consumer
+  (`WelcomeEmailDaemon`, `ResetEmailDaemon`, more to come — e.g. reports).
+- `DAEMONS` registry dict maps string name → daemon class.
+- `run_daemon.py` (project root) is the single entrypoint: `python
+  run_daemon.py <name>` runs one daemon, `python run_daemon.py --all` reads
+  `configs/daemons.json` (`{"daemons": [...]}`) and runs all enabled ones
+  concurrently via `asyncio.gather`.
+- Each daemon runs as its own Docker service/container (`command: ["python",
+  "run_daemon.py", "<name>"]`), same pattern as the `migration` service.
+- `daemons.json` is read with a plain Pydantic `BaseModel` + manual
+  `json.load`, NOT `pydantic-settings` `json_file` — that requires wiring
+  `settings_customise_sources` manually in this pydantic-settings version
+  and isn't worth the complexity here.
+
+## Docker / Poetry groups
+
+- `pyproject.toml` uses PEP 621 `[project.dependencies]` for shared deps
+  (sqlalchemy, redis, aio-pika, pydantic, bcrypt, jose, aiofiles, asyncpg,
+  psycopg2-binary, alembic, greenlet).
+- `web` group: fastapi, uvicorn, gunicorn, python-multipart — only needed
+  by the API server.
+- `daemon` group: worker-only deps (currently empty, grows as needed).
+- `dev` group: pytest stack, allure, httpie, requests-async.
+- Dockerfile has parallel builder→final stage pairs: `builder`→`prod`
+  (installs `main,web`) and `worker-builder`→`worker` (installs
+  `main,daemon`). Same base pattern: venv builder stage copies
+  `/opt/venv` into a clean final stage, poetry itself is uninstalled
+  after install to keep the final image lean.
+- Alembic runs against a **separate sync engine** (`asyncpg` swapped out,
+  psycopg2 used instead) — async SQLAlchemy engine can't drive Alembic
+  directly without the `run_sync` bridge, and a dedicated sync engine is
+  simpler than that bridge.
+- `DB_HOST` differs between contexts: `psql` (Docker service name) for
+  containers talking to each other, `localhost` for anything run on the
+  host (e.g. local `alembic revision --autogenerate`). Compose services
+  override `DB_HOST` via `environment:`; the `.env` file's own default is
+  for host-side runs.
+
+## Testing (`tests/unit`, `tests/integrated`, `tests/e2e`)
+
+- **Recurring root cause of "different event loop" / `MissingGreenlet`-style
+  errors**: prod code uses module-level singletons (`engine`, `redis_client`)
+  created once at import time and reused for the app's whole lifetime — this
+  is correct for prod (one event loop, whole uptime) but breaks under
+  pytest-asyncio's default `function`-scoped event loop (a new loop per
+  test, but the singleton's connections stay bound to the *first* loop).
+  Fix: test fixtures create a **fresh** `engine`/`RedisClient` per test and
+  monkeypatch or inject them in place of the global singleton, then dispose
+  on teardown — not a global `session`-scoped event loop (that would mask
+  real isolation bugs).
+- e2e `MySession` must subclass `httpx.AsyncClient` (not `requests_async.
+  AsyncSession` — that library silently drops cookies between requests,
+  which broke refresh-token-cookie-dependent tests like logout).
+- `test_user_fixture` is `indirect=True` parametrized with
+  `(direct_permissions, group)` tuples.

daemon_run.py

@@ -3,11 +3,13 @@ import os
 import sys
 
 from src.daemons.registry import DAEMONS
+from src.logging.logger import LogWriter
 from src.messaging.rabbitmq_client import rabbitmq_client
 from src.messaging.topology_setup import apply_topology
 from src.models.configs_read.daemons_json import daemons_config
 from src.models.rabbitmq_models.email import email_topology
 
+writer = LogWriter()
 
 async def run_one(daemon_name: str) -> None:
     daemon_cls = DAEMONS.get(daemon_name)
@@ -39,6 +41,8 @@ async def run_enabled_from_config() -> None:
 async def main() -> None:
     channel = await rabbitmq_client.get_channel()
     await apply_topology(channel, email_topology)
+    writer_task = asyncio.create_task(writer.log_writer())
+    try:
         if len(sys.argv) < 2:
             print("Usage: python run_daemon.py <daemon_name> | --all")
             sys.exit(1)
@@ -48,6 +52,8 @@ async def main() -> None:
             await run_enabled_from_config()
         else:
             await run_one(arg)
+    finally:
+        writer_task.cancel()
 
 
 if __name__ == "__main__":

main.py

@@ -6,7 +6,8 @@ from fastapi import FastAPI
 
 from src.cache.redis_client import redis_client
 from src.database.users.crud import Seed
-from src.logging.logger import LoggingMiddleware, LogWriter, ProcessingTimeMiddleware
+from src.logging.http_logger import LoggingMiddleware, ProcessingTimeMiddleware
+from src.logging.logger import LogWriter
 from src.messaging.rabbitmq_client import rabbitmq_client
 from src.messaging.topology_setup import apply_topology
 from src.models.rabbitmq_models.email import email_topology

src/logging/__init__.py

@@ -2,8 +2,14 @@
 
 import logging
 
-from .logger import LoggerDB
+from .logger import LoggerDaemon, LoggerDB
 
 sql_logger = logging.getLogger("sqlalchemy.engine")
 sql_logger.setLevel(logging.INFO)
 sql_logger.addHandler(LoggerDB())
+sql_logger.addHandler(logging.StreamHandler())
+
+daemon_logger=logging.getLogger("daemon")
+daemon_logger.setLevel(logging.INFO)
+daemon_logger.addHandler(LoggerDaemon())
+daemon_logger.addHandler(logging.StreamHandler())

src/logging/http_logger.py

@@ -0,0 +1,63 @@
+import json
+from time import perf_counter
+from typing import cast
+from uuid import uuid4
+
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from starlette.concurrency import iterate_in_threadpool
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.responses import Response, StreamingResponse
+
+from src.logging.logger import log_queue, request_id_ctx
+
+
+class ProcessingTimeMiddleware(BaseHTTPMiddleware, ):
+    async def dispatch(self, request: Request, call_next)->Response:
+        start_time = perf_counter()
+        response = await call_next(request)
+        process_time = perf_counter() - start_time
+        response.headers["X-Process-Time"] = str(process_time)
+        return response
+
+
+class LoggingMiddleware(BaseHTTPMiddleware):
+    
+    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
+            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 = []
+        async for chunk in streaming_response.body_iterator:
+            chunks.append(chunk.encode() if isinstance(chunk, str) else bytes(chunk))
+
+        body_bytes = b"".join(chunks)
+        streaming_response.body_iterator = iterate_in_threadpool(iter([body_bytes]))
+
+        try:
+            parsed = json.loads(body_bytes)
+            body = parsed.get("detail", None) if not isinstance(parsed, bool) else None
+        except (json.JSONDecodeError, TypeError):
+            body = None
+            
+        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

src/logging/logger.py

@@ -1,19 +1,12 @@
 import asyncio
-import json
 import logging
 from contextvars import ContextVar
-from time import gmtime, perf_counter, strftime
+from time import gmtime, strftime
-from typing import cast
-from uuid import uuid4
 
 import aiofiles
-from fastapi import Request
-from fastapi.responses import JSONResponse
-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="-")
+message_id_ctx: ContextVar[str] = ContextVar("message_id", default="-")
 
 log_queue=asyncio.Queue()
 
@@ -34,60 +27,17 @@ class LogWriter:
             await f.write(f"[{current_time}] {msg}\n")
             
 
-class ProcessingTimeMiddleware(BaseHTTPMiddleware, ):
+class LoggerDB(logging.Handler):
-    async def dispatch(self, request: Request, call_next)->Response:
-        start_time = perf_counter()
-        response = await call_next(request)
-        process_time = perf_counter() - start_time
-        response.headers["X-Process-Time"] = str(process_time)
-        return response
-
-
-class LoggingMiddleware(BaseHTTPMiddleware):
-    
-    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
-            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 = []
-        async for chunk in streaming_response.body_iterator:
-            chunks.append(chunk.encode() if isinstance(chunk, str) else bytes(chunk))
-
-        body_bytes = b"".join(chunks)
-        streaming_response.body_iterator = iterate_in_threadpool(iter([body_bytes]))
-
-        try:
-            parsed = json.loads(body_bytes)
-            body = parsed.get("detail", None) if not isinstance(parsed, bool) else None
-        except (json.JSONDecodeError, TypeError):
-            body = None
-            
-        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, LogWriter):
     
     def emit(self, record: logging.LogRecord) -> None:
         msg = self.format(record)
         rid=request_id_ctx.get()
         log_queue.put_nowait(("sql",f"[{rid}] {msg}"))
+
+
+class LoggerDaemon(logging.Handler):
+    
+    def emit(self, record: logging.LogRecord)->None:
+        msg= self.format(record)
+        mid=message_id_ctx.get()
+        log_queue.put_nowait(("daemon", f"[{mid}], {msg}"))

src/messaging/consumers/consumers.py

@@ -1,6 +1,9 @@
 import json
 import smtplib
+from uuid import uuid4
 
+from src.logging import daemon_logger
+from src.logging.logger import message_id_ctx
 from src.messaging.rabbitmq_client import rabbitmq_client
 from src.service.email.email_welcome import DaemonEmailSender
 
@@ -19,10 +22,10 @@ class WelcomeEmailConsumer:
 
     
     async def process_message(self, message) -> None:
-        
+        message_id_ctx.set(str(uuid4()))
         async with message.process(ignore_processed=True):
             data = json.loads(message.body)
-            print(f"Обрабатываю: {data}, метка: {message.routing_key}")
+            daemon_logger.info(f"Обрабатываю: {data}, метка: {message.routing_key}")
             try:
                 await self.daemon.send_email(data.get("email"))
             except (
@@ -31,10 +34,10 @@ class WelcomeEmailConsumer:
                 TimeoutError,
                 ConnectionRefusedError,
             ) as exc:
-                print(f"transient error, retrying: {exc!r}")
+                daemon_logger.exception(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}")
+                daemon_logger.exception(f"permanent error, sending to DLQ: {exc!r}")
                 await message.nack(requeue=False)
 
     
@@ -65,10 +68,10 @@ class ResetEmailConsumer:
 
     
     async def process_message(self, message) -> None:
-        
+        message_id_ctx.set(str(uuid4()))
         async with message.process():
             data = json.loads(message.body)
-            print(f"Обрабатываю: {data}, метка: {message.routing_key}")
+            daemon_logger.info(f"Обрабатываю: {data}, метка: {message.routing_key}")
     
     async def start_consuming(self)->None:
         

src/messaging/rabbitmq_client.py

@@ -1,3 +1,4 @@
+import asyncio
 import aio_pika
 from aio_pika.abc import AbstractChannel, AbstractRobustConnection
 
@@ -10,6 +11,8 @@ class RabbitMQClient:
         self.channel: AbstractChannel | None = None
 
     async def connect(self) -> None:
+        for attempt in range(1,6):
+            try:
                 if self.connection is None or self.connection.is_closed:
                     self.connection = await aio_pika.connect_robust(
                         host=env_settings.RABBITMQ_HOST,
@@ -17,6 +20,14 @@ class RabbitMQClient:
                         login=env_settings.RABBITMQ_LOGIN,
                         password=env_settings.RABBITMQ_PASSWORD,
                     )
+                break
+            except Exception as exc:
+                if attempt==5:
+                    raise
+                print(f"RabbitMQ not ready yet (attempt {attempt}/5): {exc!r}, retrying...")
+                await asyncio.sleep(2**attempt)
+        if self.connection is None:
+            raise RuntimeError("Failed to esablish RabbitMQ connection. Check the server!")
         if self.channel is None or self.channel.is_closed:
             self.channel = await self.connection.channel()
 

src/web/protected_routes/auth_routes.py

@@ -74,10 +74,7 @@ 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(current_user.email)
     return {"protected router": "Hello, this is a protected router"}

src/web/protected_routes/protected_user_action_routes.py

@@ -1,5 +1,6 @@
 from fastapi import APIRouter, Depends
 
+from src.messaging.producers.producers import email_producer
 from src.models.pydantic_models.model import UserCreate, UserOut, UserUpdate
 from src.service.users_crud.users_crud import CrudService, crud_service
 from src.web.protected_routes.auth_routes import require_permissions
@@ -12,6 +13,7 @@ async def get_current_user_by_email(email:str, crud:CrudService=Depends(crud_ser
 
 @router.post("/create_user")
 async def create_user(data:UserCreate, crud:CrudService=Depends(crud_service), current_user=Depends(require_permissions("admin")))->UserOut:
+    await email_producer.send_welcome_email(current_user.email)  
     return await crud.create_user(data)
 
 @router.post("/delete_user_soft")