Coverage for /usr/local/lib/python3.12/site-packages/prefect/settings/models/logging.py: 94%
34 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1from functools import partial 1a
2from pathlib import Path 1a
3from typing import Annotated, Any, ClassVar, Literal, Union 1a
5from pydantic import ( 1a
6 AliasChoices,
7 AliasPath,
8 BeforeValidator,
9 Field,
10 model_validator,
11)
12from pydantic_settings import SettingsConfigDict 1a
13from typing_extensions import Self 1a
15from prefect.settings.base import PrefectBaseSettings, build_settings_config 1a
16from prefect.types import LogLevel, validate_set_T_from_delim_string 1a
18from ._defaults import default_logging_config_path 1a
21def max_log_size_smaller_than_batch_size(values: dict[str, Any]) -> dict[str, Any]: 1a
22 """
23 Validator for settings asserting the batch size and match log size are compatible
24 """
25 if values["batch_size"] < values["max_log_size"]: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true1a
26 raise ValueError(
27 "`PREFECT_LOGGING_TO_API_MAX_LOG_SIZE` cannot be larger than"
28 " `PREFECT_LOGGING_TO_API_BATCH_SIZE`"
29 )
30 return values 1a
33class LoggingToAPISettings(PrefectBaseSettings): 1a
34 """
35 Settings for controlling logging to the API
36 """
38 model_config: ClassVar[SettingsConfigDict] = build_settings_config( 1a
39 ("logging", "to_api")
40 )
42 enabled: bool = Field( 1a
43 default=True,
44 description="If `True`, logs will be sent to the API.",
45 )
47 batch_interval: float = Field( 1a
48 default=2.0,
49 description="The number of seconds between batched writes of logs to the API.",
50 )
52 batch_size: int = Field( 1a
53 default=4_000_000,
54 description="The number of logs to batch before sending to the API.",
55 )
57 max_log_size: int = Field( 1a
58 default=1_000_000,
59 description=(
60 "The maximum size in characters for a single log. When connected to Prefect Cloud, "
61 "this value is capped at `PREFECT_CLOUD_MAX_LOG_SIZE` (default 25,000)."
62 ),
63 )
65 when_missing_flow: Literal["warn", "error", "ignore"] = Field( 1a
66 default="warn",
67 description="""
68 Controls the behavior when loggers attempt to send logs to the API handler from outside of a flow.
70 All logs sent to the API must be associated with a flow run. The API log handler can
71 only be used outside of a flow by manually providing a flow run identifier. Logs
72 that are not associated with a flow run will not be sent to the API. This setting can
73 be used to determine if a warning or error is displayed when the identifier is missing.
75 The following options are available:
77 - "warn": Log a warning message.
78 - "error": Raise an error.
79 - "ignore": Do not log a warning message or raise an error.
80 """,
81 )
83 @model_validator(mode="after") 1a
84 def emit_warnings(self) -> Self: 1a
85 """Emits warnings for misconfiguration of logging settings."""
86 values = self.model_dump() 1a
87 values = max_log_size_smaller_than_batch_size(values) 1a
88 return self 1a
91class LoggingSettings(PrefectBaseSettings): 1a
92 """
93 Settings for controlling logging behavior
94 """
96 model_config: ClassVar[SettingsConfigDict] = build_settings_config(("logging",)) 1a
98 level: LogLevel = Field( 1a
99 default="INFO",
100 description="The default logging level for Prefect loggers.",
101 )
103 config_path: Path = Field( 1a
104 default_factory=default_logging_config_path,
105 description="A path to a logging configuration file. Defaults to $PREFECT_HOME/logging.yml",
106 validation_alias=AliasChoices(
107 AliasPath("config_path"),
108 "prefect_logging_config_path",
109 "prefect_logging_settings_path",
110 ),
111 )
113 extra_loggers: Annotated[ 1a
114 Union[str, list[str], None],
115 BeforeValidator(partial(validate_set_T_from_delim_string, type_=str)),
116 ] = Field(
117 default=None,
118 description="Additional loggers to attach to Prefect logging at runtime.",
119 )
121 log_prints: bool = Field( 1a
122 default=False,
123 description="If `True`, `print` statements in flows and tasks will be redirected to the Prefect logger for the given run.",
124 )
126 colors: bool = Field( 1a
127 default=True,
128 description="If `True`, use colors in CLI output. If `False`, output will not include colors codes.",
129 )
131 markup: bool = Field( 1a
132 default=False,
133 description="""
134 Whether to interpret strings wrapped in square brackets as a style.
135 This allows styles to be conveniently added to log messages, e.g.
136 `[red]This is a red message.[/red]`. However, the downside is, if enabled,
137 strings that contain square brackets may be inaccurately interpreted and
138 lead to incomplete output, e.g.
139 `[red]This is a red message.[/red]` may be interpreted as
140 `[red]This is a red message.[/red]`.
141 """,
142 )
144 to_api: LoggingToAPISettings = Field( 1a
145 default_factory=LoggingToAPISettings,
146 description="Settings for controlling logging to the API",
147 )