File size: 2,840 Bytes
7c6ffa6 | 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 | import contextvars
import logging
import re
import uuid
# Context variable to hold the X-Request-ID for the current async task / thread context.
request_id_ctx_var: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
class RequestIDFilter(logging.Filter):
"""Logging filter that injects the current request ID into every LogRecord."""
def filter(self, record: logging.LogRecord) -> bool:
record.request_id = request_id_ctx_var.get()
return True
def sanitize_request_id(request_id: str | None) -> str:
"""Validates and sanitizes X-Request-ID. Returns a new UUID if unsafe."""
if not request_id:
return str(uuid.uuid4())
# Length limit: max 100 characters
if len(request_id) > 100:
return str(uuid.uuid4())
# Unsafe character check: allow only safe characters (alphanumeric, hyphen, underscore, dot, colon)
if not re.match(r"^[a-zA-Z0-9\-_.:]+$", request_id):
return str(uuid.uuid4())
return request_id
def setup_request_id_logging() -> None:
"""Attaches the RequestIDFilter to root and Uvicorn log handlers,
updating the format to include [%(request_id)s] after levelname.
Is fully idempotent and avoids duplicate handlers, filters, or format mutations.
"""
filter_obj = RequestIDFilter()
# We apply this filter and format adjustment to all key loggers
loggers_to_patch = ["", "uvicorn", "uvicorn.access", "uvicorn.error", "docdoe.backend"]
for logger_name in loggers_to_patch:
logger = logging.getLogger(logger_name)
# Check if RequestIDFilter is already added to the logger
has_filter = any(isinstance(f, RequestIDFilter) for f in logger.filters)
if not has_filter:
logger.addFilter(filter_obj)
for handler in logger.handlers:
# Check if RequestIDFilter is already added to the handler
has_handler_filter = any(isinstance(f, RequestIDFilter) for f in handler.filters)
if not has_handler_filter:
handler.addFilter(filter_obj)
if handler.formatter:
fmt = getattr(handler.formatter, "_fmt", None)
if fmt and "%(request_id)s" not in fmt:
if "%(levelname)s" in fmt:
new_fmt = fmt.replace("%(levelname)s", "%(levelname)s [%(request_id)s]")
else:
new_fmt = f"[%(request_id)s] {fmt}"
# Update the formatter _fmt property directly to support custom formatters (like ColoredLevelFormatter)
handler.formatter._fmt = new_fmt
if hasattr(handler.formatter, "_style"):
handler.formatter._style._fmt = new_fmt
|