Spaces:
Runtime error
Runtime error
File size: 3,043 Bytes
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 99 100 101 102 103 |
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}")
|