Source code for structum_lab.plugins.observability.observability

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

"""Structlog Integration for Structum.

This module configures the `structlog` library to provide enterprise-grade
structured logging, compliant with the `LoggerInterface` protocol.
"""

import logging
import sys
from typing import Any, cast

import structlog  # type: ignore[import-not-found]
from structum_lab.logging.interfaces import LoggerInterface

# Configuring structlog is a side-effect, usually done at module level or init.
# But providing a configure() function is cleaner for testing.



[docs]
def configure_structlog(log_level: str = "INFO", json_logs: bool = True):
    """Configures the global structlog stack."""

    processors = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
    ]

    if json_logs:
        processors.append(structlog.processors.JSONRenderer())
    else:
        processors.append(structlog.dev.ConsoleRenderer())

    structlog.configure(
        processors=processors,  # type: ignore[arg-type]
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    # Configure Standard Library Interception
    # This ensures that libraries using logging.getLogger also get routed properly
    logging.basicConfig(
        format="%(message)s",
        stream=sys.stdout,
        level=getattr(logging, log_level.upper()),
    )




[docs]
class StructuredLogger:
    """Wrapper accessible for manual instantiation if needed.

    However, the preferred way is using get_logger().
    This class is merely a compatibility shim or provider access point.
    """


[docs]
    def __init__(self, name: str) -> None:
        """Initialize the structured logger.

        Args:
            name: Logger name.
        """
        self._logger = structlog.get_logger(name)



[docs]
    def debug(self, message: str, **kwargs: Any) -> None:
        """Log a debug message."""
        self._logger.debug(message, **kwargs)



[docs]
    def info(self, message: str, **kwargs: Any) -> None:
        """Log an info message."""
        self._logger.info(message, **kwargs)



[docs]
    def warning(self, message: str, **kwargs: Any) -> None:
        """Log a warning message."""
        self._logger.warning(message, **kwargs)



[docs]
    def error(self, message: str, **kwargs: Any) -> None:
        """Log an error message."""
        self._logger.error(message, **kwargs)



[docs]
    def critical(self, message: str, **kwargs: Any) -> None:
        """Log a critical message."""
        self._logger.critical(message, **kwargs)



[docs]
    def bind(self, **kwargs: Any) -> "StructuredLogger":
        """Bind context variables to a new logger instance."""
        new_structlog = self._logger.bind(**kwargs)
        wrapper = StructuredLogger("proxy")
        wrapper._logger = new_structlog
        return wrapper





[docs]
def get_structlog_logger(name: str) -> LoggerInterface:
    """Factory function to replace structum_lab.logging.get_logger."""
    return cast(LoggerInterface, structlog.get_logger(name))