Spaces:
Runtime error
Runtime error
File size: 3,396 Bytes
a1a6d79 307cacc 1fd6030 ce323cf 9fd6e20 bb7c9a3 ce323cf 307cacc 9fd6e20 307cacc ce323cf 307cacc c21d29c 307cacc c21d29c 307cacc ce323cf c21d29c bb7c9a3 c21d29c 307cacc ce323cf c21d29c 307cacc ce323cf c21d29c 307cacc ce323cf 307cacc c21d29c 307cacc ce323cf c21d29c 1fd6030 ce323cf c21d29c ce323cf c21d29c a1a6d79 ce323cf 92e41ba a1a6d79 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 98 99 100 101 102 103 104 |
from logging import __file__ as logging_file, basicConfig, currentframe, ERROR, getLogger, Handler, INFO, LogRecord, WARNING
from loguru import logger
from os import access, getenv, W_OK
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":
if access('/data', W_OK):
logger.add(
"/data/app.log",
rotation="10 MB",
retention="1 week",
compression="zip",
format=format_string,
level=log_level,
serialize=(log_format == "json"),
)
else:
logger.warning("The log directory (/data) is not writable!")
# Intercept standard logging messages.
basicConfig(handlers=[InterceptHandler()], level=0, force=True)
# Update logging levels for some noisy libraries.
for logger_name in ("logging", "uvicorn", "uvicorn.error", "fastapi", "httpx", "pymongo"):
getLogger(logger_name).setLevel(INFO)
for logger_name in ("apscheduler"):
getLogger(logger_name).setLevel(WARNING)
getLogger().setLevel(WARNING)
logger.info(f"Logging configured with level {log_level}")
|