Spaces:
Runtime error
Runtime error
import logging | |
import sys | |
from typing import Dict, Union | |
from loguru import logger | |
from ctp_slack_bot.core.config import settings | |
class InterceptHandler(logging.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, record: logging.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 = logging.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() -> None: | |
""" | |
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. | |
""" | |
# 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 = ( | |
"<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | " | |
"<level>{level: <8}</level> | " | |
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - " | |
"<level>{message}</level>" | |
) | |
# Add console handler | |
logger.add( | |
sys.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 | |
logging.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", | |
]: | |
logging.getLogger(logger_name).setLevel(logging.INFO) | |
logger.info(f"Logging configured with level {settings.LOG_LEVEL}") | |