from dependency_injector.wiring import Provide from logging import __file__ as logging_file, basicConfig, currentframe, getLogger, Handler, INFO, LogRecord from loguru import logger from sys import stderr from typing import Self from ctp_slack_bot.containers import Container class InterceptHandler(Handler): """ Intercept standard logging messages toward Loguru. This handler intercepts all standard logging messages and redirects them to Loguru, allowing unified logging across the application. """ def emit(self: Self, record: LogRecord) -> None: # Get corresponding Loguru level if it exists try: level = logger.level(record.levelname).name except ValueError: level = record.levelno # Find caller from where the logged message originated frame, depth = currentframe(), 2 while frame and frame.f_code.co_filename == logging_file: frame = frame.f_back depth += 1 logger.opt(depth=depth, exception=record.exc_info).log( level, record.getMessage() ) def setup_logging(container: "Container") -> None: # TODO: Perhaps get rid of the container dependence since we only need two settings. """ Configure logging with Loguru. This function sets up Loguru as the main logging provider, configures the log format based on settings, and intercepts standard logging messages. """ from ctp_slack_bot.containers import Container settings = container.settings() if container else Provide[Container.settings] # Remove default loguru handler logger.remove() # Determine log format if settings.LOG_FORMAT == "json": log_format = { "time": "{time:YYYY-MM-DD HH:mm:ss.SSS}", "level": "{level}", "message": "{message}", "module": "{module}", "function": "{function}", "line": "{line}", } format_string = lambda record: record["message"] else: format_string = ( "{time:YYYY-MM-DD HH:mm:ss.SSS} | " "{level: <8} | " "{name}:{function}:{line} - " "{message}" ) # Add console handler logger.add( stderr, format=format_string, level=settings.LOG_LEVEL, serialize=(settings.LOG_FORMAT == "json"), backtrace=True, diagnose=True, ) # Add file handler for non-DEBUG environments if settings.LOG_LEVEL != "DEBUG": logger.add( "logs/app.log", rotation="10 MB", retention="1 week", compression="zip", format=format_string, level=settings.LOG_LEVEL, serialize=(settings.LOG_FORMAT == "json"), ) # Intercept standard logging messages basicConfig(handlers=[InterceptHandler()], level=0, force=True) # Update logging levels for some noisy libraries for logger_name in ("uvicorn", "uvicorn.error", "fastapi", "httpx", "apscheduler", "pymongo"): getLogger(logger_name).setLevel(INFO) logger.info(f"Logging configured with level {settings.LOG_LEVEL}")