Upgraded modules (#8)

Reviewed-on: #8
- Upgraded modules

Co-authored-by: Florian <flo@meergans.org>

src/main.py

@@ -16,14 +16,13 @@ from metrics_server import REQUEST_COUNTER
 from rabbitmq_handler import RabbitMQProducer
 from secret_manager import cleanup_secret_files
 from simple_logger_handler import LOG_LEVEL, setup_logger
-from uvicorn_logger_config import LOGGING_CONFIG
 from validator import verify_api_key
 
 logger = setup_logger(__name__)
 api_key_header_internal = APIKeyHeader(name="X-API-Key-Internal")
 
 class Notification(BaseModel):
-	user_id : int = Field(gt=0) #ToDo fix among all services
+	user_id : int = Field(gt=0)
 	message : Dict
 
 class PullResponse(BaseModel):
@@ -36,12 +35,12 @@ class PullRequest(BaseModel):
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-	logger.info("Starting application...")
+	logger.info("[FastAPI] Starting application...")
 
-	logger.info("Creating MySQL connection pool...")
+	logger.info("[FastAPI] Creating MySQL connection pool...")
 	create_connection_pool("ApiInternalPool")
 	start_healthcheck_thread()
-	logger.info("MySQL healthcheck thread started.")
+	logger.info("[FastAPI] MySQL healthcheck thread started.")
 
 	producer = RabbitMQProducer()
 	PULL_SERVICE_URL = os.getenv(
@@ -50,7 +49,7 @@ async def lifespan(app: FastAPI):
 	)
 
 	try:
-		logger.info("Starting RabbitMQ producer...")
+		logger.info("[FastAPI] Starting RabbitMQ producer...")
 		await producer.connect()
 		app.state.rmq_producer = producer
 		logger.info("[FastAPI] RabbitMQ producer initialized.")
@@ -71,10 +70,10 @@ async def lifespan(app: FastAPI):
 
 		yield
 	finally:
-		logger.info("Closing RabbitMQ producer...")
+		logger.info("[FastAPI] Closing RabbitMQ producer...")
 		await producer.close()
 
-		logger.info("Closing MySQL connection pool...")
+		logger.info("[FastAPI] Closing MySQL connection pool...")
 		close_connection_pool()
 
 		await app.state.http_client.aclose()
@@ -96,11 +95,11 @@ async def prometheus_middleware(request: Request, call_next):
 		status = response.status_code
 		logger.debug(f"Request processed with status: {status}")
 	except Exception as e:
-		logger.exception(f"Exception during request processing: {e}")
+		logger.exception(f"[FastAPI] Exception during request processing: {e}")
 		raise
 	finally:
 		REQUEST_COUNTER.labels(request.method, request.url.path, status).inc()
-		logger.debug(f"Prometheus counter incremented: {request.method} {request.url.path} {status}")
+		logger.debug(f"[FastAPI] Prometheus counter incremented: {request.method} {request.url.path} {status}")
 	return response
 
 def verify_api_key_dependency_internal(db=Depends(get_db), api_key: str = Depends(api_key_header_internal)) -> str:
@@ -109,9 +108,9 @@ def verify_api_key_dependency_internal(db=Depends(get_db), api_key: str = Depend
 		rows = cursor.fetchall()
 	for program_name, hashed_key in rows:
 		if verify_api_key(api_key=api_key, hashed=hashed_key):
-			logger.debug(f"API key verified for program: {program_name}")
+			logger.debug(f"[FastAPI] API key verified for program: {program_name}")
 			return program_name
-	logger.warning("Internal API key verification failed")
+	logger.warning("[FastAPI] Internal API key verification failed")
 	raise HTTPException(status_code=403, detail="Unauthorized")
 
 @api.exception_handler(StarletteHTTPException)
@@ -127,7 +126,7 @@ async def custom_http_exception_handler(request,exc):
 		content={"detail": exc.detail}
 	)
 
-@api.post("/internal/receive-notifications") #ToDo fix among all services
+@api.post("/internal/receive-notifications")
 async def receive_notifications(
 	notification_data: Notification,
 	request: Request,
@@ -135,8 +134,8 @@ async def receive_notifications(
 	program_name: str = Depends(verify_api_key_dependency_internal)
 ):
 
-	logger.info(f"Received notifcation data from {program_name} for RMQ")
-	logger.debug(f"Notification data: {notification_data}")
+	logger.info(f"[FastAPI] Received notifcation data from {program_name} for RMQ")
+	logger.debug(f"[FastAPI] Notification data: {notification_data}")
 	await request.app.state.rmq_producer.publish(
 		notification_data.user_id,
 		notification_data.message
@@ -168,15 +167,15 @@ async def pull_notifications(
 	return data
 
 async def start_servers():
-	logger.debug("Starting FastAPI and metrics servers")
-	config_main = uvicorn.Config("main:api", host="0.0.0.0", port=8101, log_level=LOG_LEVEL.lower(), log_config=LOGGING_CONFIG)
-	config_metrics = uvicorn.Config("metrics_server:metrics_api", host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(), log_config=LOGGING_CONFIG)
+	logger.info("[Server] Starting main API (port 8101) and metrics server (port 9000)...")
+	config_main = uvicorn.Config("main:api", host="0.0.0.0", port=8101, log_level=LOG_LEVEL.lower(), log_config=None)
+	config_metrics = uvicorn.Config("metrics_server:metrics_api", host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(), log_config=None)
 
 	server_main = uvicorn.Server(config_main)
 	server_metrics = uvicorn.Server(config_metrics)
 
 	await asyncio.gather(server_main.serve(), server_metrics.serve())
-	logger.debug("Servers started")
+	logger.info("[Server] Both servers have stopped.")
 
 if __name__ == "__main__":
 	asyncio.run(start_servers())

src/rabbitmq_handler.py

@@ -8,7 +8,7 @@ from urllib.parse import quote
 import aio_pika
 from aio_pika.exceptions import AMQPException, DeliveryError
 
-from secret_manager import return_credentials
+from secret_manager import get_secret
 from simple_logger_handler import setup_logger
 
 logger = setup_logger(__name__)
@@ -50,8 +50,8 @@ class RabbitMQProducer:
 		self._ready = asyncio.Event()
 
 	async def connect(self):
-		username = return_credentials("/etc/secrets/rmq_username")
-		password = return_credentials("/etc/secrets/rmq_password")
+		username = get_secret(os.getenv("RMQ_USERNAME",""),"/etc/secrets/rmq_username")
+		password = get_secret(os.getenv("RMQ_PASSWORD",""),"/etc/secrets/rmq_password")
 		url = (
 			f"amqp://{quote(username, safe='')}:{quote(password, safe='')}"
 			f"@{self.rmq_host}/{quote(self.rmq_vhost, safe='')}"