""" Production-standard logging configuration for AUTH Microservice. Provides structured logging with file rotation, JSON formatting, and proper log levels. """ import logging import logging.handlers import json import sys from pathlib import Path from datetime import datetime from typing import Optional from app.core.config import settings class JSONFormatter(logging.Formatter): """Custom JSON formatter for structured logging.""" def _serialize_value(self, value): """Convert non-serializable values to strings.""" try: json.dumps(value) return value except (TypeError, ValueError): return str(value) def format(self, record: logging.LogRecord) -> str: """Format log record as JSON.""" log_data = { "timestamp": datetime.utcnow().isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), "module": record.module, "function": record.funcName, "line": record.lineno, } # Add exception info if present if record.exc_info: log_data["exception"] = self.formatException(record.exc_info) # Add extra fields from the record if hasattr(record, "extra") and record.extra: for key, value in record.extra.items(): log_data[key] = self._serialize_value(value) # Add extra context from LogRecord attributes for key, value in record.__dict__.items(): if key not in [ "name", "msg", "args", "created", "filename", "funcName", "levelname", "levelno", "lineno", "module", "msecs", "message", "pathname", "process", "processName", "relativeCreated", "thread", "threadName", "exc_info", "exc_text", "stack_info", "getMessage", "extra" ]: if not key.startswith("_"): log_data[key] = self._serialize_value(value) return json.dumps(log_data, default=str) class StructuredLogger: """Structured logger wrapper for production use.""" def __init__(self, name: str): self.logger = logging.getLogger(name) def _log( self, level: int, message: str, extra: Optional[dict] = None, exc_info: bool = False, ): """Internal logging method with extra context.""" if extra: self.logger.log(level, message, extra={"extra": extra}, exc_info=exc_info) else: self.logger.log(level, message, exc_info=exc_info) def debug(self, message: str, extra: Optional[dict] = None): """Log debug message.""" self._log(logging.DEBUG, message, extra) def info(self, message: str, extra: Optional[dict] = None): """Log info message.""" self._log(logging.INFO, message, extra) def warning(self, message: str, extra: Optional[dict] = None): """Log warning message.""" self._log(logging.WARNING, message, extra) def error(self, message: str, extra: Optional[dict] = None, exc_info: bool = False): """Log error message.""" self._log(logging.ERROR, message, extra, exc_info) def critical(self, message: str, extra: Optional[dict] = None, exc_info: bool = False): """Log critical message.""" self._log(logging.CRITICAL, message, extra, exc_info) def setup_logging( log_level: str = "INFO", log_dir: str = "logs", console_json: bool = False, ) -> None: """ Configure production-standard logging. Args: log_level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) log_dir: Directory for log files console_json: Whether to output JSON to console (default: human-readable) """ # Create logs directory log_path = Path(log_dir) log_path.mkdir(exist_ok=True) # Get root logger root_logger = logging.getLogger() root_logger.setLevel(getattr(logging, log_level.upper())) # Remove existing handlers root_logger.handlers = [] # Create formatters console_formatter = ( JSONFormatter() if console_json else logging.Formatter( fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) ) json_formatter = JSONFormatter() # Console Handler (stderr) console_handler = logging.StreamHandler(sys.stderr) console_handler.setLevel(getattr(logging, log_level.upper())) console_handler.setFormatter(console_formatter) root_logger.addHandler(console_handler) # File Handler - All logs (JSON format) file_handler = logging.handlers.RotatingFileHandler( filename=log_path / "app.log", maxBytes=10 * 1024 * 1024, # 10MB backupCount=10, encoding="utf-8" ) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(json_formatter) root_logger.addHandler(file_handler) # File Handler - Info and above (JSON format) info_handler = logging.handlers.RotatingFileHandler( filename=log_path / "app_info.log", maxBytes=10 * 1024 * 1024, # 10MB backupCount=5, encoding="utf-8" ) info_handler.setLevel(logging.INFO) info_handler.setFormatter(json_formatter) root_logger.addHandler(info_handler) # File Handler - Errors only (JSON format) error_handler = logging.handlers.RotatingFileHandler( filename=log_path / "app_errors.log", maxBytes=10 * 1024 * 1024, # 10MB backupCount=10, encoding="utf-8" ) error_handler.setLevel(logging.ERROR) error_handler.setFormatter(json_formatter) root_logger.addHandler(error_handler) # Reduce noisy loggers logging.getLogger("uvicorn.access").setLevel(logging.WARNING) logging.getLogger("motor").setLevel(logging.WARNING) logging.getLogger("asyncio").setLevel(logging.WARNING) # Log startup message root_logger.info( "Logging configured", extra={ "log_level": log_level, "log_dir": str(log_path), "handlers": [type(h).__name__ for h in root_logger.handlers] } ) def get_logger(name: str) -> StructuredLogger: """ Get a structured logger instance. Args: name: Logger name (typically __name__) Returns: StructuredLogger: Configured logger instance """ return StructuredLogger(name)