Unified logging behaviour
- Logger doesn't start with log level DEBUG by default, instead reads a environment variable - Secret handler raises exceptions instead of using the module os to exit - Added extensive debug logging - Fixed RabbitMQ exception catch being too broad in scope
This commit is contained in:
1 parent
2be35613e2
commit
3758c75808
6 files changed
+56
-56
No files matched your search
+14
-7
@@ -1,5 +1,6 @@
|
||||
import asyncio
|
||||
import aio_pika
|
||||
from aio_pika.exceptions import AMQPException
|
||||
from secret_handler import return_credentials
|
||||
import os
|
||||
from logger_handler import setup_logger
|
||||
@@ -27,26 +28,30 @@ class RabbitMQProducer:
|
||||
self._flush_task: asyncio.Task | None = None
|
||||
self._closing = False
|
||||
self._ready = asyncio.Event()
|
||||
logger.debug(f"[RabbitMQ] Initialized producer for exchange '{self.exchange_name}'")
|
||||
|
||||
async def connect(self):
|
||||
logger.info(f"[RabbitMQ] Connecting to {self.url}...")
|
||||
self.connection = await aio_pika.connect_robust(self.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
|
||||
)
|
||||
logger.info(f"[aio-pika] Connected and exchange '{self.exchange_name}' ready.")
|
||||
logger.info(f"[RabbitMQ] Connected and exchange '{self.exchange_name}' 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))
|
||||
logger.debug(f"[RabbitMQ] Queued message for {routing_key}")
|
||||
|
||||
async def _flush_queue_loop(self):
|
||||
logger.debug("[RabbitMQ] Waiting for ready signal before flushing queue")
|
||||
await self._ready.wait()
|
||||
logger.debug(f"here")
|
||||
while True:
|
||||
routing_key, message = await self._queue.get()
|
||||
logger.debug(f"[RabbitMQ] Publishing message to routing key {routing_key}: {message}")
|
||||
try:
|
||||
await self.exchange.publish(
|
||||
aio_pika.Message(
|
||||
@@ -57,25 +62,27 @@ class RabbitMQProducer:
|
||||
routing_key=f"notify.user.{routing_key}",
|
||||
mandatory=True
|
||||
)
|
||||
logger.debug(f"[aio-pika] Published message to notify.user.{routing_key}")
|
||||
except Exception as e:
|
||||
logger.warning(f"[aio-pika] Publish failed, requeuing: {e}")
|
||||
logger.debug(f"[RabbitMQ] Successfully published to notify.user.{routing_key}")
|
||||
except (AMQPException, ConnectionError, OSError) as e:
|
||||
logger.warning(f"[RabbitMQ] Publish failed, requeuing message: {e}", exc_info=True)
|
||||
await asyncio.sleep(2)
|
||||
await self._queue.put((routing_key, message))
|
||||
finally:
|
||||
self._queue.task_done()
|
||||
|
||||
async def close(self):
|
||||
logger.info("[RabbitMQ] Closing producer...")
|
||||
self._closing = True
|
||||
if self._flush_task and not self._flush_task.done():
|
||||
self._flush_task.cancel()
|
||||
try:
|
||||
await self._flush_task
|
||||
logger.debug("[RabbitMQ] Flush task cancelled successfully")
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
logger.debug("[RabbitMQ] Flush task cancellation caught")
|
||||
if self.connection:
|
||||
await self.connection.close()
|
||||
logger.info("[aio-pika] Connection closed.")
|
||||
logger.info("[RabbitMQ] Connection closed")
|
||||
|
||||
|
||||
async def main():
|
||||
|
||||
Reference in new issue
Block a user