path: root/venv/lib/python3.11/site-packages/litestar/logging/config.py

from __future__ import annotations

import sys
from abc import ABC, abstractmethod
from dataclasses import asdict, dataclass, field, fields
from importlib.util import find_spec
from logging import INFO
from typing import TYPE_CHECKING, Any, Callable, Literal, cast

from litestar.exceptions import ImproperlyConfiguredException, MissingDependencyException
from litestar.serialization.msgspec_hooks import _msgspec_json_encoder
from litestar.utils.deprecation import deprecated

__all__ = ("BaseLoggingConfig", "LoggingConfig", "StructLoggingConfig")


if TYPE_CHECKING:
    from collections.abc import Iterable
    from typing import NoReturn

    # these imports are duplicated on purpose so sphinx autodoc can find and link them
    from structlog.types import BindableLogger, Processor, WrappedLogger
    from structlog.typing import EventDict

    from litestar.types import Logger, Scope
    from litestar.types.callable_types import ExceptionLoggingHandler, GetLogger


try:
    from structlog.types import BindableLogger, Processor, WrappedLogger
except ImportError:
    BindableLogger = Any  # type: ignore[assignment, misc]
    Processor = Any  # type: ignore[misc]
    WrappedLogger = Any  # type: ignore[misc]


default_handlers: dict[str, dict[str, Any]] = {
    "console": {
        "class": "logging.StreamHandler",
        "level": "DEBUG",
        "formatter": "standard",
    },
    "queue_listener": {
        "class": "litestar.logging.standard.QueueListenerHandler",
        "level": "DEBUG",
        "formatter": "standard",
    },
}

if sys.version_info >= (3, 12, 0):
    default_handlers["queue_listener"].update(
        {
            "class": "logging.handlers.QueueHandler",
            "queue": {
                "()": "queue.Queue",
                "maxsize": -1,
            },
            "listener": "litestar.logging.standard.LoggingQueueListener",
            "handlers": ["console"],
        }
    )

    # do not format twice, the console handler will do the job
    del default_handlers["queue_listener"]["formatter"]


default_picologging_handlers: dict[str, dict[str, Any]] = {
    "console": {
        "class": "picologging.StreamHandler",
        "level": "DEBUG",
        "formatter": "standard",
    },
    "queue_listener": {
        "class": "litestar.logging.picologging.QueueListenerHandler",
        "level": "DEBUG",
        "formatter": "standard",
    },
}


def get_logger_placeholder(_: str | None = None) -> NoReturn:
    """Raise: An :class:`ImproperlyConfiguredException <.exceptions.ImproperlyConfiguredException>`"""
    raise ImproperlyConfiguredException(
        "cannot call '.get_logger' without passing 'logging_config' to the Litestar constructor first"
    )


def _get_default_handlers() -> dict[str, dict[str, Any]]:
    """Return the default logging handlers for the config.

    Returns:
        A dictionary of logging handlers
    """
    if find_spec("picologging"):
        return default_picologging_handlers
    return default_handlers


def _default_exception_logging_handler_factory(
    is_struct_logger: bool, traceback_line_limit: int
) -> ExceptionLoggingHandler:
    """Create an exception logging handler function.

    Args:
        is_struct_logger: Whether the logger is a structlog instance.
        traceback_line_limit: Maximal number of lines to log from the
            traceback.

    Returns:
        An exception logging handler.
    """

    def _default_exception_logging_handler(logger: Logger, scope: Scope, tb: list[str]) -> None:
        # we limit the length of the stack trace to 20 lines.
        first_line = tb.pop(0)

        if is_struct_logger:
            logger.exception(
                "Uncaught Exception",
                connection_type=scope["type"],
                path=scope["path"],
                traceback="".join(tb[-traceback_line_limit:]),
            )
        else:
            stack_trace = first_line + "".join(tb[-traceback_line_limit:])
            logger.exception(
                "exception raised on %s connection to route %s\n\n%s", scope["type"], scope["path"], stack_trace
            )

    return _default_exception_logging_handler


class BaseLoggingConfig(ABC):
    """Abstract class that should be extended by logging configs."""

    __slots__ = ("log_exceptions", "traceback_line_limit", "exception_logging_handler")

    log_exceptions: Literal["always", "debug", "never"]
    """Should exceptions be logged, defaults to log exceptions when ``app.debug == True``'"""
    traceback_line_limit: int
    """Max number of lines to print for exception traceback"""
    exception_logging_handler: ExceptionLoggingHandler | None
    """Handler function for logging exceptions."""

    @abstractmethod
    def configure(self) -> GetLogger:
        """Return logger with the given configuration.

        Returns:
            A 'logging.getLogger' like function.
        """
        raise NotImplementedError("abstract method")

    @staticmethod
    def set_level(logger: Any, level: int) -> None:
        """Provides a consistent interface to call `setLevel` for all loggers."""
        raise NotImplementedError("abstract method")


@dataclass
class LoggingConfig(BaseLoggingConfig):
    """Configuration class for standard logging.

    Notes:
        - If 'picologging' is installed it will be used by default.
    """

    version: Literal[1] = field(default=1)
    """The only valid value at present is 1."""
    incremental: bool = field(default=False)
    """Whether the configuration is to be interpreted as incremental to the existing configuration.

    Notes:
        - This option is ignored for 'picologging'
    """
    disable_existing_loggers: bool = field(default=False)
    """Whether any existing non-root loggers are to be disabled."""
    filters: dict[str, dict[str, Any]] | None = field(default=None)
    """A dict in which each key is a filter id and each value is a dict describing how to configure the corresponding
    Filter instance.
    """
    propagate: bool = field(default=True)
    """If messages must propagate to handlers higher up the logger hierarchy from this logger."""
    formatters: dict[str, dict[str, Any]] = field(
        default_factory=lambda: {
            "standard": {"format": "%(levelname)s - %(asctime)s - %(name)s - %(module)s - %(message)s"}
        }
    )
    handlers: dict[str, dict[str, Any]] = field(default_factory=_get_default_handlers)
    """A dict in which each key is a handler id and each value is a dict describing how to configure the corresponding
    Handler instance.
    """
    loggers: dict[str, dict[str, Any]] = field(
        default_factory=lambda: {
            "litestar": {"level": "INFO", "handlers": ["queue_listener"], "propagate": False},
        }
    )
    """A dict in which each key is a logger name and each value is a dict describing how to configure the corresponding
    Logger instance.
    """
    root: dict[str, dict[str, Any] | list[Any] | str] = field(
        default_factory=lambda: {
            "handlers": ["queue_listener"],
            "level": "INFO",
        }
    )
    """This will be the configuration for the root logger.

    Processing of the configuration will be as for any logger, except that the propagate setting will not be applicable.
    """
    configure_root_logger: bool = field(default=True)
    """Should the root logger be configured, defaults to True for ease of configuration."""
    log_exceptions: Literal["always", "debug", "never"] = field(default="debug")
    """Should exceptions be logged, defaults to log exceptions when 'app.debug == True'"""
    traceback_line_limit: int = field(default=20)
    """Max number of lines to print for exception traceback"""
    exception_logging_handler: ExceptionLoggingHandler | None = field(default=None)
    """Handler function for logging exceptions."""

    def __post_init__(self) -> None:
        if "queue_listener" not in self.handlers:
            self.handlers["queue_listener"] = _get_default_handlers()["queue_listener"]

        if "litestar" not in self.loggers:
            self.loggers["litestar"] = {
                "level": "INFO",
                "handlers": ["queue_listener"],
                "propagate": False,
            }

        if self.log_exceptions != "never" and self.exception_logging_handler is None:
            self.exception_logging_handler = _default_exception_logging_handler_factory(
                is_struct_logger=False, traceback_line_limit=self.traceback_line_limit
            )

    def configure(self) -> GetLogger:
        """Return logger with the given configuration.

        Returns:
            A 'logging.getLogger' like function.
        """

        excluded_fields: tuple[str, ...]
        if "picologging" in " ".join([handler["class"] for handler in self.handlers.values()]):
            try:
                from picologging import config, getLogger
            except ImportError as e:
                raise MissingDependencyException("picologging") from e

            excluded_fields = ("incremental", "configure_root_logger")
        else:
            from logging import config, getLogger  # type: ignore[no-redef, assignment]

            excluded_fields = ("configure_root_logger",)

        values = {
            _field.name: getattr(self, _field.name)
            for _field in fields(self)
            if getattr(self, _field.name) is not None and _field.name not in excluded_fields
        }

        if not self.configure_root_logger:
            values.pop("root")
        config.dictConfig(values)
        return cast("Callable[[str], Logger]", getLogger)

    @staticmethod
    def set_level(logger: Logger, level: int) -> None:
        """Provides a consistent interface to call `setLevel` for all loggers."""
        logger.setLevel(level)


class StructlogEventFilter:
    """Remove keys from the log event.

    Add an instance to the processor chain.

    .. code-block:: python
        :caption: Examples

        structlog.configure(
            ...,
            processors=[
                ...,
                EventFilter(["color_message"]),
                ...,
            ],
        )

    """

    def __init__(self, filter_keys: Iterable[str]) -> None:
        """Initialize the EventFilter.

        Args:
            filter_keys: Iterable of string keys to be excluded from the log event.
        """
        self.filter_keys = filter_keys

    def __call__(self, _: WrappedLogger, __: str, event_dict: EventDict) -> EventDict:
        """Receive the log event, and filter keys.

        Args:
            _ ():
            __ ():
            event_dict (): The data to be logged.

        Returns:
            The log event with any key in `self.filter_keys` removed.
        """
        for key in self.filter_keys:
            event_dict.pop(key, None)
        return event_dict


def default_json_serializer(value: EventDict, **_: Any) -> bytes:
    return _msgspec_json_encoder.encode(value)


def stdlib_json_serializer(value: EventDict, **_: Any) -> str:  # pragma: no cover
    return _msgspec_json_encoder.encode(value).decode("utf-8")


def default_structlog_processors(as_json: bool = True) -> list[Processor]:  # pyright: ignore
    """Set the default processors for structlog.

    Returns:
        An optional list of processors.
    """
    try:
        import structlog
        from structlog.dev import RichTracebackFormatter

        if as_json:
            return [
                structlog.contextvars.merge_contextvars,
                structlog.processors.add_log_level,
                structlog.processors.format_exc_info,
                structlog.processors.TimeStamper(fmt="iso"),
                structlog.processors.JSONRenderer(serializer=default_json_serializer),
            ]
        return [
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.dev.ConsoleRenderer(
                colors=True, exception_formatter=RichTracebackFormatter(max_frames=1, show_locals=False, width=80)
            ),
        ]

    except ImportError:
        return []


def default_structlog_standard_lib_processors(as_json: bool = True) -> list[Processor]:  # pyright: ignore
    """Set the default processors for structlog stdlib.

    Returns:
        An optional list of processors.
    """
    try:
        import structlog
        from structlog.dev import RichTracebackFormatter

        if as_json:
            return [
                structlog.processors.TimeStamper(fmt="iso"),
                structlog.stdlib.add_log_level,
                structlog.stdlib.ExtraAdder(),
                StructlogEventFilter(["color_message"]),
                structlog.stdlib.ProcessorFormatter.remove_processors_meta,
                structlog.processors.JSONRenderer(serializer=stdlib_json_serializer),
            ]
        return [
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.stdlib.add_log_level,
            structlog.stdlib.ExtraAdder(),
            StructlogEventFilter(["color_message"]),
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            structlog.dev.ConsoleRenderer(
                colors=True, exception_formatter=RichTracebackFormatter(max_frames=1, show_locals=False, width=80)
            ),
        ]
    except ImportError:
        return []


def default_logger_factory(as_json: bool = True) -> Callable[..., WrappedLogger] | None:
    """Set the default logger factory for structlog.

    Returns:
        An optional logger factory.
    """
    try:
        import structlog

        if as_json:
            return structlog.BytesLoggerFactory()
        return structlog.WriteLoggerFactory()
    except ImportError:
        return None


@dataclass
class StructLoggingConfig(BaseLoggingConfig):
    """Configuration class for structlog.

    Notes:
        - requires ``structlog`` to be installed.
    """

    processors: list[Processor] | None = field(default=None)  # pyright: ignore
    """Iterable of structlog logging processors."""
    standard_lib_logging_config: LoggingConfig | None = field(default=None)  # pyright: ignore
    """Optional customized standard logging configuration.

    Use this when you need to modify the standard library outside of the Structlog pre-configured implementation.
    """
    wrapper_class: type[BindableLogger] | None = field(default=None)  # pyright: ignore
    """Structlog bindable logger."""
    context_class: dict[str, Any] | None = None
    """Context class (a 'contextvar' context) for the logger."""
    logger_factory: Callable[..., WrappedLogger] | None = field(default=None)  # pyright: ignore
    """Logger factory to use."""
    cache_logger_on_first_use: bool = field(default=True)
    """Whether to cache the logger configuration and reuse."""
    log_exceptions: Literal["always", "debug", "never"] = field(default="debug")
    """Should exceptions be logged, defaults to log exceptions when 'app.debug == True'"""
    traceback_line_limit: int = field(default=20)
    """Max number of lines to print for exception traceback"""
    exception_logging_handler: ExceptionLoggingHandler | None = field(default=None)
    """Handler function for logging exceptions."""
    pretty_print_tty: bool = field(default=True)
    """Pretty print log output when run from an interactive terminal."""

    def __post_init__(self) -> None:
        if self.processors is None:
            self.processors = default_structlog_processors(not sys.stderr.isatty() and self.pretty_print_tty)
        if self.logger_factory is None:
            self.logger_factory = default_logger_factory(not sys.stderr.isatty() and self.pretty_print_tty)
        if self.log_exceptions != "never" and self.exception_logging_handler is None:
            self.exception_logging_handler = _default_exception_logging_handler_factory(
                is_struct_logger=True, traceback_line_limit=self.traceback_line_limit
            )
        try:
            import structlog

            if self.standard_lib_logging_config is None:
                self.standard_lib_logging_config = LoggingConfig(
                    formatters={
                        "standard": {
                            "()": structlog.stdlib.ProcessorFormatter,
                            "processors": default_structlog_standard_lib_processors(
                                as_json=not sys.stderr.isatty() and self.pretty_print_tty
                            ),
                        }
                    }
                )
        except ImportError:
            self.standard_lib_logging_config = LoggingConfig()

    def configure(self) -> GetLogger:
        """Return logger with the given configuration.

        Returns:
            A 'logging.getLogger' like function.
        """
        try:
            import structlog
        except ImportError as e:
            raise MissingDependencyException("structlog") from e

        structlog.configure(
            **{
                k: v
                for k, v in asdict(self).items()
                if k
                not in (
                    "standard_lib_logging_config",
                    "log_exceptions",
                    "traceback_line_limit",
                    "exception_logging_handler",
                    "pretty_print_tty",
                )
            }
        )
        return structlog.get_logger

    @staticmethod
    def set_level(logger: Logger, level: int) -> None:
        """Provides a consistent interface to call `setLevel` for all loggers."""

        try:
            import structlog

            structlog.configure(wrapper_class=structlog.make_filtering_bound_logger(level))
        except ImportError:
            """"""
            return


@deprecated(version="2.6.0", removal_in="3.0.0", alternative="`StructLoggingConfig.set_level`")
def default_wrapper_class(log_level: int = INFO) -> type[BindableLogger] | None:  # pragma: no cover  # pyright: ignore
    try:  # pragma: no cover
        import structlog

        return structlog.make_filtering_bound_logger(log_level)
    except ImportError:
        return None