app/utils/request_id.py

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