""" Logging configuration for MTG Online backend. Provides structured logging with support for both text and JSON output formats. Configured via environment variables or the application config. ## Configuration | Variable | Default | Description | |---|---|---| | LOG_LEVEL | INFO | Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) | | LOG_FORMAT | text | Output format ("text" or "json") | | MTG_LOG_LEVEL | INFO | Alternative env var for log level | | MTG_LOG_FORMAT | text | Alternative env var for log format | ## Usage from app.core.logging import configure_logging # Auto-configures on import, or call explicitly: configure_logging() ## Features - Service name "mtgonline-backend" added to all log records - Container name from HOSTNAME env var (for Docker) - Suppresses noisy SQLAlchemy loggers unless DEBUG - JSON format includes timestamp, level, service, container, logger, message - Text format: `[timestamp] [service] LEVEL - message` """ import json import logging import os import sys from datetime import datetime, timezone from typing import Optional def _get_container_name() -> str: """Get container name from HOSTNAME environment variable. Returns: Container name or "unknown" if not available. """ return os.environ.get("HOSTNAME", "unknown") class ContainerFormatter(logging.Formatter): """Formatter that adds container metadata to log records. Supports both text and JSON output formats. Adds service name "mtgonline-backend" and container name to all log records. Attributes: SERVICE_NAME: Service identifier added to all log records. fmt_type: Output format ("text" or "json"). container_name: Container identifier from HOSTNAME env var. """ SERVICE_NAME = "mtgonline-backend" def __init__(self, fmt_type: str = "text") -> None: """Initialize the formatter. Args: fmt_type: Output format, either "text" or "json". """ super().__init__() self.fmt_type = fmt_type self.container_name = _get_container_name() def format(self, record: logging.LogRecord) -> str: """Format a log record with service and container metadata. Args: record: The log record to format. Returns: Formatted log string in the configured format. """ record.service = self.SERVICE_NAME record.container = self.container_name if self.fmt_type == "json": return self._format_json(record) return self._format_text(record) def _format_text(self, record: logging.LogRecord) -> str: """Format log record as human-readable text. Args: record: The log record to format. Returns: Text-formatted log string. """ return ( f"{self.formatTime(record)} [{record.service}] " f"{record.levelname} - {record.getMessage()}" ) def _format_json(self, record: logging.LogRecord) -> str: """Format log record as JSON. Includes timestamp, level, service, container, logger, and message. Adds exception information if present. Args: record: The log record to format. Returns: JSON-formatted log string. """ log_entry = { "timestamp": datetime.fromtimestamp( record.created, tz=timezone.utc ).isoformat(), "level": record.levelname, "service": record.service, "container": record.container, "logger": record.name, "message": record.getMessage(), } if record.exc_info and record.exc_info[0] is not None: log_entry["exception"] = self.formatException(record.exc_info) return json.dumps(log_entry, default=str) def configure_logging( log_level: Optional[str] = None, log_format: Optional[str] = None, ) -> None: """Configure application logging. Reads configuration from environment variables or uses defaults. Sets up a stream handler with the configured formatter. Suppresses noisy SQLAlchemy loggers unless DEBUG level is enabled. Args: log_level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Overrides MTG_LOG_LEVEL and LOG_FORMAT env vars if provided. log_format: Output format ("text" or "json"). Overrides MTG_LOG_FORMAT and LOG_FORMAT env vars if provided. Example: >>> configure_logging() # Use env vars or defaults >>> configure_logging(log_level="DEBUG", log_format="json") """ # Determine log level if log_level is None: log_level = os.environ.get("MTG_LOG_LEVEL", os.environ.get("LOG_LEVEL", "INFO")).upper() # Determine log format if log_format is None: log_format = os.environ.get("MTG_LOG_FORMAT", os.environ.get("LOG_FORMAT", "text")).lower() # Validate inputs level = getattr(logging, log_level, logging.INFO) if log_format not in ("text", "json"): log_format = "text" # Create formatter and handler formatter = ContainerFormatter(fmt_type=log_format) handler = logging.StreamHandler(sys.stdout) handler.setFormatter(formatter) # Configure root logger root_logger = logging.getLogger() root_logger.setLevel(level) root_logger.handlers.clear() root_logger.addHandler(handler) # Suppress noisy loggers unless DEBUG if level != logging.DEBUG: logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING) logging.getLogger("sqlalchemy.pool").setLevel(logging.WARNING) logger = logging.getLogger(__name__) logger.info( f"Logging configured: level={log_level}, " f"format={log_format}, container={_get_container_name()}" ) # Auto-configure on import configure_logging()