Coverage for /usr/local/lib/python3.12/site-packages/prefect/_internal/observability.py: 57%
24 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1"""
2internal module for configuring observability tooling (logfire, etc.)
3"""
5from typing import Any 1a
7from pydantic import Field 1a
8from pydantic_settings import BaseSettings, SettingsConfigDict 1a
11class LogfireSettings(BaseSettings): 1a
12 """
13 configuration for logfire observability integration.
14 """
16 model_config = SettingsConfigDict( 1a
17 env_prefix="PREFECT_LOGFIRE_",
18 extra="ignore",
19 )
21 enabled: bool = Field( 1a
22 default=False,
23 description="whether to enable logfire observability",
24 )
26 write_token: str | None = Field( 1a
27 default=None,
28 description="API token for writing to logfire. required when enabled=true.",
29 )
31 sampling_head_rate: float = Field( 1a
32 default=0.1,
33 ge=0.0,
34 le=1.0,
35 description="fraction of traces to sample upfront (0.0-1.0). reduces total volume.",
36 )
38 sampling_level_threshold: str = Field( 1a
39 default="warn",
40 description="minimum log level to always include (debug, info, warn, error). keeps all warnings/errors.",
41 )
43 sampling_duration_threshold: float = Field( 1a
44 default=5.0,
45 ge=0.0,
46 description="minimum duration in seconds to always include. catches slow operations.",
47 )
49 sampling_background_rate: float = Field( 1a
50 default=0.01,
51 ge=0.0,
52 le=1.0,
53 description="fraction of non-notable traces to keep anyway (0.0-1.0). maintains baseline visibility.",
54 )
57def configure_logfire() -> Any | None: 1a
58 """
59 configure and return logfire instance with sampling, or None if disabled.
61 this function:
62 1. checks if logfire is enabled via PREFECT_LOGFIRE_ENABLED
63 2. validates PREFECT_LOGFIRE_WRITE_TOKEN is set
64 3. loads sampling configuration from env vars
65 4. configures logfire with sampling options
66 5. returns configured logfire module (or None if disabled)
68 can be called multiple times safely - logfire.configure is idempotent.
69 """
70 # load logfire settings from env vars
71 settings = LogfireSettings() 1a
73 if not settings.enabled: 73 ↛ 76line 73 didn't jump to line 76 because the condition on line 73 was always true1a
74 return None 1a
76 if settings.write_token is None:
77 raise ValueError(
78 "PREFECT_LOGFIRE_WRITE_TOKEN must be set when PREFECT_LOGFIRE_ENABLED is true"
79 )
81 try:
82 import logfire # pyright: ignore
83 except ImportError as exc:
84 raise ImportError(
85 "logfire is not installed. install it with: uv add logfire"
86 ) from exc
88 # build sampling options
89 sampling_options = logfire.SamplingOptions.level_or_duration(
90 head=settings.sampling_head_rate,
91 level_threshold=settings.sampling_level_threshold,
92 duration_threshold=settings.sampling_duration_threshold,
93 background_rate=settings.sampling_background_rate,
94 )
96 logfire.configure(token=settings.write_token, sampling=sampling_options) # pyright: ignore
97 return logfire