""" Structured logging per PecFlow. In produzione (LOG_JSON=true) emette log JSON per aggregatori (Loki, ELK). In sviluppo emette log leggibili colorati. """ import logging import sys from typing import Any from app.config import get_settings settings = get_settings() def _build_handler() -> logging.Handler: handler = logging.StreamHandler(sys.stdout) if settings.log_json: try: import json class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: log_entry: dict[str, Any] = { "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"), "level": record.levelname, "logger": record.name, "message": record.getMessage(), } if record.exc_info: log_entry["exception"] = self.formatException(record.exc_info) return json.dumps(log_entry, ensure_ascii=False) handler.setFormatter(JsonFormatter()) except Exception: pass else: fmt = "%(asctime)s %(levelname)-8s %(name)s – %(message)s" handler.setFormatter(logging.Formatter(fmt, datefmt="%H:%M:%S")) return handler def setup_logging() -> None: """Configura il logging applicativo. Da chiamare all'avvio dell'app.""" level = getattr(logging, settings.log_level.upper(), logging.INFO) root_logger = logging.getLogger() root_logger.setLevel(level) # Rimuovi handler esistenti per evitare duplicati root_logger.handlers.clear() root_logger.addHandler(_build_handler()) # Riduci verbosità librerie rumorose logging.getLogger("uvicorn.access").setLevel(logging.WARNING) logging.getLogger("sqlalchemy.engine").setLevel( logging.INFO if settings.app_debug else logging.WARNING ) def get_logger(name: str) -> logging.Logger: """Restituisce un logger con il nome specificato.""" return logging.getLogger(name)