Source code for structum_lab.logging

# src/structum_lab/logging/__init__.py
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods

"""
Logging and Metrics interface for Structum.
Provides fallback implementations that can be replaced by plugins.
"""

import logging
from typing import Any, Optional, Protocol, Dict

# --- Logger Abstraction ---

from .interfaces import LoggerInterface



[docs]
class StandardLoggerAdapter:
    """Fallback implementation that adapts standard logging.Logger to LoggerInterface.

    It captures structured arguments (**kwargs) and places them into the 'extra'
    dictionary, creating a rudimentary structured logging experience even without
    advanced plugins.
    """


[docs]
    def __init__(self, logger: logging.Logger):
        self.logger = logger
        self._reserved_keys = {"exc_info", "stack_info", "stacklevel", "extra"}


    def _log(self, level_method, message: str, **kwargs: Any) -> None:
        # Extract standard logging arguments
        log_kwargs = {k: v for k, v in kwargs.items() if k in self._reserved_keys}
        # Everything else goes into 'extra'
        extra = {k: v for k, v in kwargs.items() if k not in self._reserved_keys}

        # If 'extra' was explicitly passed, merge it (though it's rare in this usage)
        if "extra" in kwargs:
            extra.update(kwargs["extra"])

        level_method(message, extra=extra, **log_kwargs)


[docs]
    def debug(self, message: str, **kwargs: Any) -> None:
        self._log(self.logger.debug, message, **kwargs)



[docs]
    def info(self, message: str, **kwargs: Any) -> None:
        self._log(self.logger.info, message, **kwargs)



[docs]
    def warning(self, message: str, **kwargs: Any) -> None:
        self._log(self.logger.warning, message, **kwargs)



[docs]
    def error(self, message: str, **kwargs: Any) -> None:
        self._log(self.logger.error, message, **kwargs)



[docs]
    def critical(self, message: str, **kwargs: Any) -> None:
        self._log(self.logger.critical, message, **kwargs)





[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).
    """
    return StandardLoggerAdapter(logging.getLogger(name))



# --- Metrics Abstraction ---



[docs]
class MetricsCollectorProtocol(Protocol):

[docs]
    def increment(

        self, metric: str, labels: Optional[Dict[str, str]] = None
    ) -> None: ...

[docs]
    def observe(

        self, metric: str, value: float, labels: Optional[Dict[str, str]] = None
    ) -> None: ...




[docs]
class NullMetrics:
    """Implementazione 'Null Object' per le metriche (fallback)."""


[docs]
    def increment(self, metric: str, labels: Optional[Dict[str, str]] = None) -> None:
        pass



[docs]
    def observe(
        self, metric: str, value: float, labels: Optional[Dict[str, str]] = None
    ) -> None:
        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 configure_logging(level: str = "INFO", format: str = "json") -> None:
    """
    Combines basic logging configuration.
    Plugins should patch this to provide advanced setup.
    """
    logging.basicConfig(
        level=level, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )




[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)
from contextlib import contextmanager



[docs]
def set_context(**kwargs: Any) -> None:
    """Sets global context variables (Fallback: No-op)."""
    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)."""
    yield