From 06adc919bd4a7ff924a111ec80675a888b68beff Mon Sep 17 00:00:00 2001 From: Florian Date: Sat, 10 Oct 2026 13:11:06 +0200 Subject: [PATCH] Added Windows routing key and moved environment variables into init section --- src/main.py | 37 +++++++++++++++++++----------------- src/rabbitmq_handler.py | 42 +++++++++++++++++++++-------------------- 2 files changed, 42 insertions(+), 37 deletions(-) diff --git a/src/main.py b/src/main.py index a7f6df3..145f07c 100644 --- a/src/main.py +++ b/src/main.py @@ -16,7 +16,6 @@ from uvicorn_logger_config import LOGGING_CONFIG from secret_manager import cleanup_secret_files logger = setup_logger(__name__) -producer = RabbitMQProducer() api_key_header_internal = APIKeyHeader(name="X-API-Key-Internal") class Notification(BaseModel): @@ -34,25 +33,29 @@ async def lifespan(app: FastAPI): start_healthcheck_thread() logger.info("MySQL healthcheck thread started.") - logger.info("Starting RabbitMQ producer...") - await producer.connect() - app.state.rmq_producer = producer - logger.info("[FastAPI] RabbitMQ producer initialized.") + producer = RabbitMQProducer() - SECRET_PATHS = frozenset({ - "/etc/secrets/db_username", - "/etc/secrets/db_password", - "/etc/secrets/rmq_username", - "/etc/secrets/rmq_password" - }) - cleanup_secret_files(SECRET_PATHS) + try: + logger.info("Starting RabbitMQ producer...") + await producer.connect() + app.state.rmq_producer = producer + logger.info("[FastAPI] RabbitMQ producer initialized.") - yield - logger.info("Closing RabbitMQ producer...") - await producer.close() + SECRET_PATHS = frozenset({ + "/etc/secrets/db_username", + "/etc/secrets/db_password", + "/etc/secrets/rmq_username", + "/etc/secrets/rmq_password" + }) + cleanup_secret_files(SECRET_PATHS) - logger.info("Closing MySQL connection pool...") - close_connection_pool() + yield + finally: + logger.info("Closing RabbitMQ producer...") + await producer.close() + + logger.info("Closing MySQL connection pool...") + close_connection_pool() api = FastAPI( title="Internal Notifier API", diff --git a/src/rabbitmq_handler.py b/src/rabbitmq_handler.py index 9819480..eb93117 100644 --- a/src/rabbitmq_handler.py +++ b/src/rabbitmq_handler.py @@ -5,45 +5,46 @@ from secret_manager import return_credentials import os from simple_logger_handler import setup_logger import json +from urllib.parse import quote logger = setup_logger(__name__) -rmq_username = return_credentials("/etc/secrets/rmq_username") -rmq_password = return_credentials("/etc/secrets/rmq_password") -rmq_host = os.getenv("BACKEND_API_INTERNAL_RMQ_HOST", "localhost") -rmq_vhost = os.getenv("BACKEND_API_INTERNAL_RMQ_VHOST", "app_notifications") -rmq_exchange = os.getenv("BACKEND_API_INTERNAL_RMQ_EXCHANGE", "app_notifications") - -RABBITMQ_URL = f"amqp://{rmq_username}:{rmq_password}@{rmq_host}/{rmq_vhost}" - - class RabbitMQProducer: - def __init__(self, url=RABBITMQ_URL, exchange_name=rmq_exchange,rmq_host=rmq_host): - self.url = url - self.rmq_host = rmq_host - self.exchange_name = exchange_name + def __init__(self): + self.rmq_host = os.getenv("BACKEND_API_INTERNAL_RMQ_HOST", "localhost") + self.rmq_vhost = os.getenv("BACKEND_API_INTERNAL_RMQ_VHOST", "app_notifications") + self.rmq_exchange = os.getenv("BACKEND_API_INTERNAL_RMQ_EXCHANGE", "app_notifications") + self.connection: aio_pika.RobustConnection | None = None self.channel: aio_pika.RobustChannel | None = None self.exchange: aio_pika.Exchange | None = None - self._queue: asyncio.Queue[tuple[int, dict]] = asyncio.Queue() + self._queue: asyncio.Queue[tuple[str, dict]] = asyncio.Queue() self._flush_task: asyncio.Task | None = None self._closing = False self._ready = asyncio.Event() async def connect(self): + username = return_credentials("/etc/secrets/rmq_username") + password = return_credentials("/etc/secrets/rmq_password") + url = ( + f"amqp://{quote(username, safe='')}:{quote(password, safe='')}" + f"@{self.rmq_host}/{quote(self.rmq_vhost, safe='')}" + ) + logger.info(f"[RabbitMQ] Connecting to {self.rmq_host}...") - self.connection = await aio_pika.connect_robust(self.url) + self.connection = await aio_pika.connect_robust(url) self.channel = await self.connection.channel(publisher_confirms=True) self.exchange = await self.channel.declare_exchange( - self.exchange_name, aio_pika.ExchangeType.TOPIC, durable=True + self.rmq_exchange, aio_pika.ExchangeType.TOPIC, durable=True ) - logger.info(f"[RabbitMQ] Connected and exchange '{self.exchange_name}' ready.") + logger.info(f"[RabbitMQ] Connected and exchange '{self.rmq_exchange}' ready.") self._ready.set() self._flush_task = asyncio.create_task(self._flush_queue_loop()) logger.debug("[RabbitMQ] Flush queue loop started") async def publish(self, routing_key: int, message: dict): - await self._queue.put((routing_key, message)) + await self._queue.put((f"notify.user.{routing_key}", message)) + await self._queue.put((f"notify.windows.{routing_key}", message)) logger.debug(f"[RabbitMQ] Queued message for {routing_key}") async def _flush_queue_loop(self): @@ -59,12 +60,13 @@ class RabbitMQProducer: content_type="application/json", delivery_mode=aio_pika.DeliveryMode.PERSISTENT ), - routing_key=f"notify.user.{routing_key}", + routing_key=routing_key, mandatory=True ) logger.debug(f"[RabbitMQ] Successfully published to notify.user.{routing_key}") + except (AMQPException, OSError) as e: - logger.warning(f"[RabbitMQ] Publish failed, requeuing message: {e}", exc_info=True) + logger.warning(f"[RabbitMQ] Failed to publish to {routing_key}: {e}. Requeuing message.", exc_info=True) await asyncio.sleep(2) await self._queue.put((routing_key, message)) finally: