Spaces:
Runtime error
Runtime error
File size: 3,089 Bytes
ce323cf 307cacc c21d29c ce323cf 9fd6e20 ce323cf 307cacc 9fd6e20 307cacc ce323cf 307cacc c21d29c 307cacc c21d29c 307cacc ce323cf c21d29c 307cacc ce323cf c21d29c 307cacc ce323cf c21d29c 307cacc ce323cf 307cacc c21d29c 307cacc ce323cf c21d29c 307cacc d3e6eca 307cacc c21d29c 307cacc ce323cf c21d29c ce323cf c21d29c ce323cf c21d29c |
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 |
from logging import __file__ as logging_file, basicConfig, currentframe, getLogger, Handler, INFO, LogRecord
from loguru import logger
from os import getenv
from sys import stderr
from typing import Self
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() -> None:
"""
Configure logging with Loguru.
This function sets up Loguru as the main logging provider, configures the log format based on environment variables,
and intercepts standard logging messages.
"""
# Get logger configuration from environment variables.
log_level = getenv("LOG_LEVEL", "INFO")
log_format = getenv("LOG_FORMAT", "text")
# Remove default loguru handler.
logger.remove()
# Determine log format.
if 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=log_level,
serialize=(log_format == "json"),
backtrace=True,
diagnose=True,
)
# Add file handler for non-DEBUG environments.
if log_level != "DEBUG":
logger.add(
"/data/app.log",
rotation="10 MB",
retention="1 week",
compression="zip",
format=format_string,
level=log_level,
serialize=(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 {log_level}")
|