blob: fafa3dde8f1b3ba4bc0b35aa7533c26b5fec9d56 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
from litestar.cli._utils import console
from litestar.logging.config import StructLoggingConfig
from litestar.middleware.logging import LoggingMiddlewareConfig
from litestar.plugins import InitPluginProtocol
if TYPE_CHECKING:
from litestar.config.app import AppConfig
@dataclass
class StructlogConfig:
structlog_logging_config: StructLoggingConfig = field(default_factory=StructLoggingConfig)
"""Structlog Logging configuration for Litestar. See ``litestar.logging.config.StructLoggingConfig``` for details."""
middleware_logging_config: LoggingMiddlewareConfig = field(default_factory=LoggingMiddlewareConfig)
"""Middleware logging config."""
enable_middleware_logging: bool = True
"""Enable request logging."""
class StructlogPlugin(InitPluginProtocol):
"""Structlog Plugin."""
__slots__ = ("_config",)
def __init__(self, config: StructlogConfig | None = None) -> None:
if config is None:
config = StructlogConfig()
self._config = config
super().__init__()
def on_app_init(self, app_config: AppConfig) -> AppConfig:
"""Structlog Plugin
Args:
app_config: The :class:`AppConfig <litestar.config.app.AppConfig>` instance.
Returns:
The app config object.
"""
if app_config.logging_config is not None and isinstance(app_config.logging_config, StructLoggingConfig):
console.print(
"[red dim]* Found pre-configured `StructLoggingConfig` on the `app` instance. Skipping configuration.[/]",
)
else:
app_config.logging_config = self._config.structlog_logging_config
app_config.logging_config.configure()
if self._config.structlog_logging_config.standard_lib_logging_config is not None: # pragma: no cover
self._config.structlog_logging_config.standard_lib_logging_config.configure() # pragma: no cover
if self._config.enable_middleware_logging:
app_config.middleware.append(self._config.middleware_logging_config.middleware)
return app_config # pragma: no cover
|
