# src/structum/logging/__init__.py
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods
"""
Logging and Metrics interface for Structum.
This module acts as the **Architectural Facade** (DP-1) for observability.
It provides:
1. **Unified API**: A consistent interface for logging and metrics regardless of the backend.
2. **Operational Continuity**: Fallback implementations (Adapters & Null Objects) that ensure the application runs even without plugins.
3. **Zero Dependency**: Pure stdlib implementations by default.
Plugins (like `structum-observability`) patch these objects at runtime to inject advanced behavior.
"""
import logging
from collections.abc import Callable
from contextlib import contextmanager
from typing import Any, Protocol
# --- Logger Abstraction ---
from .interfaces import LoggerInterface
[docs]
class StandardLoggerAdapter:
"""Fallback implementation that adapts standard logging.Logger to LoggerInterface.
**Architectural Role:** Operational Continuity (DP-2).
This class ensures that calls to `log.info(..., user_id=123)` do not crash
when the advanced observability plugin is missing. It captures structured
arguments (**kwargs) and places them into the 'extra' dictionary, creating
a rudimentary structured logging experience compatible with standard tools.
"""
[docs]
def __init__(self, logger: logging.Logger) -> None:
"""Initialize the adapter with a stdlib logger.
Args:
logger: The standard library Logger instance to wrap.
"""
self.logger = logger
self._reserved_keys = {"exc_info", "stack_info", "stacklevel", "extra"}
def _log(self, level_method, message: str, **kwargs: Any) -> None:
"""Internal method to log with structured extras.
Args:
level_method: The logging method (e.g., logger.info).
message: The log message.
**kwargs: Structured key-value pairs to include in the log.
"""
log_kwargs = {k: v for k, v in kwargs.items() if k in self._reserved_keys}
extra = {k: v for k, v in kwargs.items() if k not in self._reserved_keys}
if "extra" in kwargs:
extra.update(kwargs["extra"])
level_method(message, extra=extra, **log_kwargs)
[docs]
def debug(self, message: str, **kwargs: Any) -> None:
"""Log a debug message."""
self._log(self.logger.debug, message, **kwargs)
[docs]
def info(self, message: str, **kwargs: Any) -> None:
"""Log an info message."""
self._log(self.logger.info, message, **kwargs)
[docs]
def warning(self, message: str, **kwargs: Any) -> None:
"""Log a warning message."""
self._log(self.logger.warning, message, **kwargs)
[docs]
def error(self, message: str, **kwargs: Any) -> None:
"""Log an error message."""
self._log(self.logger.error, message, **kwargs)
[docs]
def critical(self, message: str, **kwargs: Any) -> None:
"""Log a critical message."""
self._log(self.logger.critical, message, **kwargs)
_logger_factory: Callable[[str], LoggerInterface] | None = None
[docs]
def get_logger(name: str) -> LoggerInterface:
"""Returns a logger instance conforming to LoggerInterface.
By default, this returns a StandardLoggerAdapter wrapping the stdlib logger.
Plugins (like structum_observability) should patch this function to return
their own implementation (e.g., a structlog BoundLogger).
"""
if _logger_factory:
return _logger_factory(name)
return StandardLoggerAdapter(logging.getLogger(name))
# --- Metrics Abstraction ---
[docs]
class MetricsCollectorProtocol(Protocol):
"""Protocol defining the metrics collection interface.
Implementations should provide methods for incrementing counters
and observing values (e.g., for histograms or gauges).
"""
[docs]
def increment(self, metric: str, labels: dict[str, str] | None = None) -> None:
"""Increment a counter metric."""
...
[docs]
def observe(self, metric: str, value: float, labels: dict[str, str] | None = None) -> None:
"""Observe a value for a histogram or gauge metric."""
...
[docs]
class NullMetrics:
"""Null Object implementation for metrics (fallback when no backend is configured).
**Architectural Role**: Operational Continuity (DP-2).
This class implements the `MetricsCollectorProtocol` but performs no actions.
It allows application code to instrumentation itself (`metrics.increment(...)`)
without checking if a metrics backend is actually installed.
"""
[docs]
def increment(self, metric: str, labels: dict[str, str] | None = None) -> None:
"""No-op increment. Swallows the metric to prevent runtime errors."""
pass
[docs]
def observe(self, metric: str, value: float, labels: dict[str, str] | None = None) -> None:
"""No-op observe. Swallows the value to prevent runtime errors."""
pass
# Default global instance (fallback)
# Default global instance (fallback)
metrics: MetricsCollectorProtocol = NullMetrics()
[docs]
def set_metrics_collector(collector: MetricsCollectorProtocol) -> None:
"""Permette ai plugin di registrare il proprio collettore di metriche."""
global metrics
metrics = collector
# --- Configuration & Context Abstraction ---
[docs]
def get_logger_backend() -> str:
"""Returns the name of the active logging backend."""
return "stdlib"
# Context vars stubs (Plugins will override these with real context propagation)
[docs]
def set_context(**kwargs: Any) -> None:
"""Sets global context variables (Fallback: No-op).
In the Core implementation, this is a **No-Op**.
Context propagation requires the `structum-observability` plugin
which implements `contextvars` management.
"""
pass
[docs]
def clear_context() -> None:
"""Clears global context variables (Fallback: No-op)."""
pass
[docs]
@contextmanager
def bind_context(**kwargs: Any):
"""Context manager for temporary context (Fallback: yield).
Without the plugin, this simply yields control back to the caller
without modifying any context, ensuring code compatibility.
"""
yield