""" Logging configuration for NLProxy. Features: - Environment-aware configuration (dev vs production) - Structured JSON logging for production (machine-readable, ELK/Datadog ready) - Pretty/colored logging for development (human-readable) - Request/context binding via thread-local contextual filters - Log rotation, filtering, and multi-handler support - Thread-safe and async-compatible Usage: >>> from nlproxy.utils.logger import setup_logging, get_request_logger >>> setup_logging(level="INFO") >>> logger = get_request_logger("proxy") >>> ContextFilter.set_context(request_id="abc123") >>> logger.info("Request processed") Author: IntelliDeep Labs Team License: BSL 1.1 """ from __future__ import annotations import json import logging import os import sys import threading from datetime import datetime, timezone from logging.handlers import RotatingFileHandler from typing import Any, Dict, Optional # ============================================================================= # CONTEXT FILTER (Binds request_id, trace_id, etc. to all log records) # ============================================================================= class ContextFilter(logging.Filter): """ Injects contextual fields (request_id, user_id, etc.) into log records. Thread-safe via thread-local storage. Compatible with FastAPI/Starlette middleware. """ _context: threading.local = threading.local() def filter(self, record: logging.LogRecord) -> bool: ctx = getattr(self._context, "data", {}) for key, value in ctx.items(): setattr(record, key, value) return True @classmethod def set_context(cls, **kwargs: Any) -> None: """Set context variables for the current thread.""" cls._context.data = kwargs @classmethod def get_context(cls) -> Dict[str, Any]: """Retrieve current thread context.""" return getattr(cls._context, "data", {}) @classmethod def clear_context(cls) -> None: """Clear thread context (call at request teardown).""" cls._context.data = {} # ============================================================================= # FORMATTERS # ============================================================================= class JSONFormatter(logging.Formatter): """ Production-ready JSON formatter for structured logging. Outputs parseable JSON objects compatible with observability platforms. """ def format(self, record: logging.LogRecord) -> str: log_obj = { "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), "module": record.module, "function": record.funcName, "line": record.lineno, } # Inject context fields for key, value in record.__dict__.items(): if key not in log_obj and not key.startswith("_"): log_obj[key] = value # Handle exceptions if record.exc_info and record.exc_info[0] is not None: log_obj["exception"] = self.formatException(record.exc_info) return json.dumps(log_obj, default=str, ensure_ascii=False) class PrettyFormatter(logging.Formatter): """ Human-readable formatter for development/debugging. Includes ANSI colors if supported by the terminal. """ COLORS = { logging.DEBUG: "\033[36m", logging.INFO: "\033[32m", logging.WARNING: "\033[33m", logging.ERROR: "\033[31m", logging.CRITICAL: "\033[1;31m", } RESET = "\033[0m" def format(self, record: logging.LogRecord) -> str: color = self.COLORS.get(record.levelno, self.RESET) ctx = f" [{record.request_id}]" if hasattr(record, "request_id") else "" ts = datetime.fromtimestamp(record.created, tz=timezone.utc).strftime("%H:%M:%S") return ( f"{color}[{ts}][{record.levelname}]{self.RESET} " f"{record.name}{ctx}: {record.getMessage()}" ) # ============================================================================= # CORE SETUP # ============================================================================= _initialized = False _root_logger = logging.getLogger("nlproxy") def setup_logging( level: str = "INFO", format_type: str = "auto", log_dir: Optional[str] = None, max_bytes: int = 50 * 1024 * 1024, # 50MB backup_count: int = 5, disable_existing: bool = True, ) -> None: """ Initialize logging for the nlproxy application. Parameters ---------- level : str Log level: DEBUG, INFO, WARNING, ERROR, CRITICAL. format_type : str "json" for production, "pretty" for dev, "auto" detects environment. log_dir : Optional[str] Directory for log files. If None, only console logging is used. max_bytes : int Max size per log file before rotation. backup_count : int Number of rotated log files to keep. disable_existing : bool Whether to clear handlers from third-party libraries. """ global _initialized if _initialized: return numeric_level = getattr(logging, level.upper(), logging.INFO) env = os.getenv("NLPROXY_ENV", "development").lower() if format_type == "auto": format_type = "json" if env in ("production", "prod", "staging") else "pretty" # Configure root logger _root_logger.setLevel(numeric_level) _root_logger.propagate = False if disable_existing: for handler in logging.root.handlers[:]: logging.root.removeHandler(handler) # Add context filter to root context_filter = ContextFilter() _root_logger.addFilter(context_filter) # Console handler console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(numeric_level) console_handler.setFormatter(JSONFormatter() if format_type == "json" else PrettyFormatter()) _root_logger.addHandler(console_handler) # File handler (optional) if log_dir: os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, f"nlproxy_{env}.log") file_handler = RotatingFileHandler( log_file, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8" ) file_handler.setLevel(numeric_level) file_handler.setFormatter(JSONFormatter()) # Always JSON for files _root_logger.addHandler(file_handler) # Silence noisy third-party loggers noisy_loggers = ["urllib3", "httpx", "httpcore", "asyncio", "watchfiles", "uvicorn.access"] for name in noisy_loggers: logging.getLogger(name).setLevel(logging.WARNING) _initialized = True _root_logger.info(f"Logging initialized: level={level}, format={format_type}, env={env}") def get_request_logger(name: str) -> logging.LoggerAdapter: """ Returns a logger adapter bound to request context. Automatically injects request_id, user_id, or any context set via ContextFilter.set_context(). Parameters ---------- name : str Logger name (typically module __name__ or "proxy", "service", etc.). Returns ------- logging.LoggerAdapter Context-aware logger. """ base_logger = logging.getLogger(f"nlproxy.{name}") return logging.LoggerAdapter(base_logger, extra=ContextFilter.get_context())