logging.py•1.62 kB
"""生产级日志配置"""
import sys
import logging
from typing import Any, Dict
import structlog
from structlog import get_logger
from src.config.settings import settings
def setup_logging() -> None:
"""设置结构化日志"""
# 配置标准库日志
logging.basicConfig(
format="%(message)s",
stream=sys.stdout,
level=getattr(logging, settings.logging.level.upper())
)
# 配置structlog
processors = [
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="ISO"),
structlog.processors.CallsiteParameterAdder(
{
structlog.processors.CallsiteParameter.FILENAME,
structlog.processors.CallsiteParameter.FUNC_NAME,
structlog.processors.CallsiteParameter.LINENO,
}
),
]
if settings.logging.format == "json":
processors.append(structlog.processors.JSONRenderer())
else:
processors.extend([
structlog.dev.ConsoleRenderer(colors=True),
])
structlog.configure(
processors=processors,
wrapper_class=structlog.make_filtering_bound_logger(
getattr(logging, settings.logging.level.upper())
),
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
def get_structured_logger(name: str = None) -> structlog.BoundLogger:
"""获取结构化日志记录器"""
return get_logger(name)
# 通用日志记录器
logger = get_structured_logger(__name__)