Coverage for src / gitversioned / logging.py: 91%
77 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-14 20:55 +0000
1"""
2Loguru-based logging configuration and environment settings for GitVersioned.
4This module provides a unified interface for configuring application-level logging
5using loguru and Pydantic settings. It handles dynamic OpenTelemetry formatting,
6sink resolution, and configurable log levels to ensure consistent telemetry
7across the codebase and build environments.
8"""
10from __future__ import annotations
12import contextlib
13import json
14import sys
15import traceback
16from collections.abc import Callable
17from typing import Any, ClassVar, Literal
19from loguru import logger
20from pydantic import Field, field_validator
21from pydantic_settings import BaseSettings, SettingsConfigDict
23from gitversioned.compat import opentelemetry_trace
25__all__ = ["LoggingSettings", "configure_logger", "logger"]
27_GITVERSIONED_HANDLER_ID: int | None = None
30class LoggingSettings(BaseSettings):
31 """
32 Settings model for configuring the loguru logging infrastructure.
34 This Pydantic model loads configuration from environment variables prefixed
35 with ``GITVERSIONED__LOGGING__`` and provides typed fields for controlling
36 log output, formatting, and OpenTelemetry integration.
38 Example:
39 .. code-block:: python
41 from gitversioned.logging import LoggingSettings, configure_logger
43 settings = LoggingSettings(enabled=True, level="DEBUG")
44 configure_logger(settings)
45 """
47 enabled: bool = Field(
48 default=False,
49 description=(
50 "Whether to enable GitVersioned loguru logging across the application."
51 ),
52 )
53 clear_loggers: bool = Field(
54 default=False,
55 description=(
56 "If true, removes all existing loguru handlers before configuring new ones."
57 ),
58 )
59 sink: str | Any = Field(
60 default=sys.stdout,
61 description=(
62 "The output sink for log messages. Can be an object or string "
63 "alias ('stdout')."
64 ),
65 )
66 level: str = Field(
67 default="INFO",
68 description="The minimum severity level for emitted log messages.",
69 )
70 otel_formatting: Literal["auto", "enable", "disable"] = Field(
71 default="auto",
72 description=(
73 "Controls OpenTelemetry JSON formatting. 'auto' enables it if "
74 "otel is installed."
75 ),
76 )
77 format: str | Callable[..., Any] | None = Field(
78 default="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | "
79 "<cyan>{name}</cyan>:<cyan>{function}</cyan> - <level>{message}</level>\n",
80 description=(
81 "The log format string or function to use when otel formatting is disabled."
82 ),
83 )
84 filter: Any = Field(
85 default=True,
86 description=(
87 "Filters log records. Defaults to True to filter by the "
88 "'gitversioned' prefix."
89 ),
90 )
91 enqueue: bool = Field(
92 default=True,
93 description="Whether to enable thread-safe asynchronous logging.",
94 )
95 kwargs: dict[str, Any] = Field(
96 default_factory=dict,
97 description=(
98 "Additional keyword arguments to pass directly to loguru's add() method."
99 ),
100 )
102 model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(
103 env_prefix="GITVERSIONED__LOGGING__",
104 env_nested_delimiter="__",
105 )
106 """Pydantic configuration dict dictating environment variable prefixes."""
108 @field_validator("sink", mode="before")
109 @classmethod
110 def _parse_sink(cls, value: Any) -> Any:
111 """Convert string aliases for stdout/stderr to actual objects."""
112 if isinstance(value, str):
113 mapping = {
114 "stdout": sys.stdout,
115 "sys.stdout": sys.stdout,
116 "stderr": sys.stderr,
117 "sys.stderr": sys.stderr,
118 }
119 return mapping.get(value.lower(), value)
120 return value
123def _otel_formatter(record: dict[str, Any]) -> str:
124 """Format the log record as an OpenTelemetry compliant JSON string."""
125 trace_id = span_id = trace_flags = None
127 if opentelemetry_trace:
128 span = opentelemetry_trace.get_current_span()
129 context = span.get_span_context()
130 if context.is_valid:
131 trace_id = format(context.trace_id, "032x")
132 span_id = format(context.span_id, "016x")
133 trace_flags = format(context.trace_flags, "02x")
135 log_record = {
136 "timestamp": record["time"].isoformat(),
137 "severity_text": record["level"].name,
138 "body": record["message"],
139 "resource": {"service.name": "gitversioned"},
140 "attributes": {
141 "module": record["name"],
142 "function": record["function"],
143 "line": record["line"],
144 "process_id": record["process"].id,
145 **record["extra"],
146 },
147 }
149 if record.get("exception"):
150 exception = record["exception"]
151 log_record["attributes"]["exception.type"] = exception.type.__name__
152 log_record["attributes"]["exception.message"] = str(exception.value)
153 log_record["attributes"]["exception.stacktrace"] = "".join(
154 traceback.format_exception(
155 exception.type, exception.value, exception.traceback
156 )
157 )
159 if trace_id:
160 log_record.update(
161 {
162 "trace_id": trace_id,
163 "span_id": span_id,
164 "trace_flags": trace_flags,
165 }
166 )
168 # Escape braces so loguru doesn't interpret the JSON string as a format string
169 return json.dumps(log_record).replace("{", "{{").replace("}", "}}") + "\n"
172def configure_logger(settings: LoggingSettings | None = None) -> None:
173 """
174 Initializes the loguru logger with the provided settings or from the environment.
176 This function configures the global loguru logger instance based on the provided
177 ``LoggingSettings``. It handles enabling/disabling the logger, managing sinks,
178 and injecting the appropriate formatter (including OpenTelemetry).
180 Example:
181 .. code-block:: python
183 from gitversioned.logging import configure_logger, LoggingSettings
185 configure_logger(LoggingSettings(level="DEBUG"))
187 :param settings: An optional instance of ``LoggingSettings``. If not provided,
188 settings are automatically loaded from the environment.
189 :return: None
190 :raises ImportError: If OpenTelemetry formatting is explicitly enabled but the
191 package is not installed.
192 """
193 global _GITVERSIONED_HANDLER_ID # noqa: PLW0603
195 settings = settings or LoggingSettings()
197 if not settings.enabled:
198 logger.disable("gitversioned")
199 return
201 logger.enable("gitversioned")
203 if settings.clear_loggers:
204 logger.remove()
205 _GITVERSIONED_HANDLER_ID = None
206 elif isinstance(_GITVERSIONED_HANDLER_ID, int):
207 with contextlib.suppress(ValueError):
208 logger.remove(_GITVERSIONED_HANDLER_ID)
209 _GITVERSIONED_HANDLER_ID = None
211 use_otel = settings.otel_formatting == "enable" or (
212 settings.otel_formatting == "auto" and opentelemetry_trace is not None
213 )
214 if settings.otel_formatting == "enable" and opentelemetry_trace is None:
215 raise ImportError(
216 "OpenTelemetry is not installed but 'otel_formatting' was set to 'enable'."
217 )
219 log_format = _otel_formatter if use_otel else settings.format
220 filter_val = "gitversioned" if settings.filter is True else settings.filter
222 if isinstance(filter_val, (list, tuple)):
223 prefixes = tuple(filter_val)
225 def final_filter(record: dict[str, Any]) -> bool:
226 return bool(record["name"] and record["name"].startswith(prefixes))
228 elif isinstance(filter_val, str):
230 def final_filter(record: dict[str, Any]) -> bool:
231 return bool(record["name"] and record["name"].startswith(filter_val))
233 else:
234 final_filter = None if filter_val is False else filter_val # type: ignore[assignment]
236 _GITVERSIONED_HANDLER_ID = logger.add(
237 settings.sink, # type: ignore[arg-type]
238 level=settings.level,
239 filter=final_filter, # type: ignore[arg-type]
240 format=log_format, # type: ignore[arg-type]
241 enqueue=settings.enqueue,
242 **settings.kwargs,
243 )