disdantic.logging¶

class InterceptHandler(logging.Handler):
    """
    Standard logging handler to intercept and forward records to Loguru.

    This class hooks into the standard Python `logging` module. When a
    standard logging record is emitted, it translates the logging level
    and redirects the message, caller context, and exception trace to the
    Loguru pipeline, ensuring unified log aggregation.

    Example:
        .. code-block:: python

            import logging
            from disdantic.logging import InterceptHandler

            logging.basicConfig(handlers=[InterceptHandler()], level=0)

    """

    def emit(self, record: logging.LogRecord) -> None:
        """
        Emit a standard library logging record by forwarding it to Loguru.

        :param record: The standard library log record.
        :returns: None.
        """
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Pass the caller info directly to loguru via patch and bind
        # to avoid slow frame walking or runtime function/closure definition.
        logger.patch(_static_patcher).bind(
            pathname=record.pathname,
            lineno=record.lineno,
            funcName=record.funcName,
        ).opt(exception=record.exc_info).log(level, record.getMessage())

class LoggingSettings(BaseSettings):
    """
    Settings configuration for the disdantic logging subsystem.

    This class defines the configuration schema for logging, loading parameters
    from environment variables prefixed with `DISDANTIC__LOGGING__` or via
    direct instantiation. It allows customizing the output sink, log level,
    format template, OpenTelemetry integration, and thread-safe queueing.

    Example:
        .. code-block:: python

            from disdantic.logging import LoggingSettings, configure_logger

            settings = LoggingSettings(
                enabled=True,
                level="DEBUG",
                sink="stdout"
            )
            configure_logger(settings)

    :cvar model_config: Configuration dictionary dictating environment variable
                        prefixes and nested delimiters.
    """

    enabled: bool = Field(
        default=False,
        description=(
            "Enables or disables logging output across the disdantic package."
        ),
    )
    clear_loggers: bool = Field(
        default=False,
        description=(
            "Configures whether all existing active logger sinks "
            "are removed prior to setup."
        ),
    )
    sink: str | Any = Field(
        default=sys.stderr,
        description=(
            "Specifies and maps the output target, such as standard streams "
            "(stdout, stderr) or a file path, for log messages."
        ),
    )
    level: str = Field(
        default="WARNING",
        description=(
            "Configures the minimum severity level required for log messages "
            "to be emitted."
        ),
    )
    otel_formatting: Literal["auto", "enable", "disable"] = Field(
        default="auto",
        description=(
            "Configures the OpenTelemetry-compliant JSON formatting option "
            "(auto, enable, or disable)."
        ),
    )
    format: str | Callable[..., Any] | None = Field(
        default="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | "
        "<cyan>{name}</cyan>:<cyan>{function}</cyan> - <level>{message}</level>\n",
        description=(
            "Configures the standard text template layout for emitted log lines."
        ),
    )
    filter: Any = Field(
        default=True,
        description=(
            "Configures the filtering criteria, using a prefix string, "
            "list of prefixes, or a filter function."
        ),
    )
    enqueue: bool = Field(
        default=True,
        description="Enables or disables asynchronous, thread-safe message queueing.",
    )
    kwargs: dict[str, Any] = Field(
        default_factory=dict,
        description=(
            "Maps additional custom arguments passed directly "
            "to the loguru add handler."
        ),
    )

    model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(
        env_prefix="DISDANTIC__LOGGING__",
        env_nested_delimiter="__",
    )

    @field_validator("sink", mode="before")
    @classmethod
    def _parse_sink(cls, value: Any) -> Any:
        # Convert string aliases for standard output/error streams to stream objects.
        if isinstance(value, str):
            mapping = {
                "stdout": sys.stdout,
                "sys.stdout": sys.stdout,
                "stderr": sys.stderr,
                "sys.stderr": sys.stderr,
            }
            return mapping.get(value.lower(), value)
        return value

class OtelSink:
    """
    Wrapper sink for OpenTelemetry-compliant JSON logging.

    This class intercepts log messages emitted by Loguru, extracts metadata
    from the Loguru record (such as exception tracebacks, trace context,
    and process info), and serializes them into standard OpenTelemetry JSON format.

    Example:
        .. code-block:: python

            import sys
            from disdantic.logging import OtelSink

            sink = OtelSink(sys.stderr)
            sink.write("Hello log message")

    """

    target: Annotated[
        Any,
        "The target output stream or file path where log records are written.",
    ]

    def __init__(self, target: Any) -> None:
        """
        Initialize the OpenTelemetry sink wrapper.

        :param target: Output stream or file path where log records are written.
        :returns: None.
        """
        self.target = target
        self._file = None
        if isinstance(target, (str, Path)):
            self._file = Path(target).open("a", encoding="utf-8")  # noqa: SIM115

    def write(self, message: str) -> None:
        """
        Write a message to the target output after formatting it as
        OpenTelemetry JSON if possible.

        :param message: The log message to serialize or write.
        :returns: None.
        """
        record = getattr(message, "record", None)
        if record is not None:
            log_record = _build_otel_record(record)
            serialized = json.dumps(log_record) + "\n"
        else:
            serialized = message

        if self._file is not None:
            self._file.write(serialized)
            self._file.flush()
        elif hasattr(self.target, "write"):
            self.target.write(serialized)
            if hasattr(self.target, "flush"):
                self.target.flush()

    def close(self) -> None:
        """
        Close the underlying file descriptor if a file path was provided as target.

        :returns: None.
        """
        if self._file is not None:
            self._file.close()