111 lines
2.9 KiB
Python
111 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import sys
|
|
|
|
from loguru import logger
|
|
|
|
# Периодические пробы мониторинга/оркестратора и служебные doc-эндпоинты.
|
|
NOISE_PATHS: tuple[str, ...] = (
|
|
"/healthcheck",
|
|
"/uptime",
|
|
"/metrics",
|
|
"/docs",
|
|
"/redoc",
|
|
"/openapi.json",
|
|
"/favicon.ico",
|
|
)
|
|
|
|
# Маршруты статики (mount'ы StaticFiles) — контейнеры любой статики.
|
|
_NOISE_MARKERS: tuple[str, ...] = ("/static/", "/static-swagger/")
|
|
|
|
# Расширения статических ассетов (картинки, стили, скрипты, шрифты).
|
|
_STATIC_SUFFIXES: tuple[str, ...] = (
|
|
".png",
|
|
".jpg",
|
|
".jpeg",
|
|
".gif",
|
|
".svg",
|
|
".ico",
|
|
".webp",
|
|
".css",
|
|
".js",
|
|
".map",
|
|
".woff",
|
|
".woff2",
|
|
".ttf",
|
|
".eot",
|
|
)
|
|
|
|
|
|
def is_noise_path(path: str) -> bool:
|
|
"""True для служебных путей, чьи УСПЕШНЫЕ обращения не логируем.
|
|
|
|
Пробы (healthcheck/uptime/metrics), doc-эндпоинты и статика — чистый шум.
|
|
Ошибки (>=400) по этим путям логируются: это решает RequestLoggingMiddleware.
|
|
"""
|
|
if any(path.endswith(p) for p in NOISE_PATHS):
|
|
return True
|
|
if any(marker in path for marker in _NOISE_MARKERS):
|
|
return True
|
|
if any(path.endswith(suffix) for suffix in _STATIC_SUFFIXES):
|
|
return True
|
|
return False
|
|
|
|
|
|
def setup_logger(
|
|
*,
|
|
level: str | int = "INFO",
|
|
fmt: str | None = None,
|
|
intercept_std_logging: bool = True,
|
|
) -> None:
|
|
logger.remove()
|
|
logger.configure(extra={"request_id": "-"})
|
|
|
|
if fmt is None:
|
|
fmt = (
|
|
"{time:YYYY-MM-DD HH:mm:ss.SSS} | "
|
|
"{level:<8} | "
|
|
"req={extra[request_id]} | "
|
|
"{name}:{function}:{line} - {message}"
|
|
)
|
|
|
|
logger.add(
|
|
sys.stdout,
|
|
level=level,
|
|
format=fmt,
|
|
enqueue=True,
|
|
backtrace=False,
|
|
diagnose=False,
|
|
)
|
|
|
|
if not intercept_std_logging:
|
|
return
|
|
|
|
class InterceptHandler(logging.Handler):
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
lvl: str | int
|
|
try:
|
|
lvl = logger.level(record.levelname).name
|
|
except ValueError:
|
|
lvl = record.levelno
|
|
|
|
logger.opt(
|
|
depth=6,
|
|
exception=record.exc_info,
|
|
).log(lvl, record.getMessage())
|
|
|
|
logging.root.handlers = [InterceptHandler()]
|
|
logging.root.setLevel(0)
|
|
|
|
for name in ("uvicorn", "uvicorn.error"):
|
|
lg = logging.getLogger(name)
|
|
lg.handlers = [InterceptHandler()]
|
|
lg.setLevel(0)
|
|
lg.propagate = False
|
|
|
|
access = logging.getLogger("uvicorn.access")
|
|
access.handlers = []
|
|
access.propagate = False
|
|
access.disabled = True
|