Spaces:
Runtime error
Runtime error
File size: 3,299 Bytes
9fd6e20 ce323cf 307cacc ce323cf 9fd6e20 307cacc ce323cf 307cacc 9fd6e20 307cacc ce323cf 307cacc 9fd6e20 307cacc 9fd6e20 ce323cf 307cacc ce323cf 307cacc ce323cf 307cacc ce323cf 307cacc ce323cf 307cacc ce323cf 307cacc ce323cf 307cacc ce323cf 307cacc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
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 = (
"<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(
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}")
|