Coverage for /usr/local/lib/python3.12/site-packages/prefect/cli/config.py: 15%
140 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1"""
2Command line interface for working with profiles
3"""
5from __future__ import annotations 1a
7import os 1a
8from pathlib import Path 1a
9from typing import Any, Union, cast 1a
11import toml 1a
12import typer 1a
13from dotenv import dotenv_values 1a
14from typing_extensions import Literal 1a
16import prefect.context 1a
17import prefect.settings 1a
18from prefect.cli._types import PrefectTyper 1a
19from prefect.cli._utilities import exit_with_error, exit_with_success 1a
20from prefect.cli.root import app, is_interactive 1a
21from prefect.exceptions import ProfileSettingsValidationError 1a
22from prefect.settings.legacy import ( 1a
23 Setting,
24 _get_settings_fields, # type: ignore[reportPrivateUsage] Private util that needs to live next to Setting class
25 _get_valid_setting_names, # type: ignore[reportPrivateUsage] Private util that needs to live next to Setting class
26)
27from prefect.utilities.annotations import NotSet 1a
28from prefect.utilities.collections import listrepr 1a
30help_message = """ 1a
31 View and set Prefect profiles.
32"""
33VALID_SETTING_NAMES = _get_valid_setting_names(prefect.settings.Settings) 1a
34config_app: PrefectTyper = PrefectTyper(name="config", help=help_message) 1a
35app.add_typer(config_app) 1a
38@config_app.command("set") 1a
39def set_(settings: list[str]): 1a
40 """
41 Change the value for a setting by setting the value in the current profile.
42 """
43 parsed_settings: dict[Union[str, Setting], Any] = {}
44 for item in settings:
45 try:
46 setting, value = item.split("=", maxsplit=1)
47 except ValueError:
48 exit_with_error(
49 f"Failed to parse argument {item!r}. Use the format 'VAR=VAL'."
50 )
52 if setting not in VALID_SETTING_NAMES:
53 exit_with_error(f"Unknown setting name {setting!r}.")
55 # Guard against changing settings that tweak config locations
56 if setting in {"PREFECT_HOME", "PREFECT_PROFILES_PATH"}:
57 exit_with_error(
58 f"Setting {setting!r} cannot be changed with this command. "
59 "Use an environment variable instead."
60 )
62 parsed_settings[setting] = value
64 try:
65 new_profile = prefect.settings.update_current_profile(parsed_settings)
66 except ProfileSettingsValidationError as exc:
67 help_message = ""
68 for setting, problem in exc.errors:
69 for error in problem.errors():
70 help_message += f"[bold red]Validation error(s) for setting[/bold red] [blue]{setting.name}[/blue]\n\n - {error['msg']}\n\n"
71 exit_with_error(help_message)
73 for setting, value in parsed_settings.items():
74 app.console.print(f"Set {setting!r} to {value!r}.")
75 if setting in os.environ:
76 app.console.print(
77 f"[yellow]{setting} is also set by an environment variable which will "
78 f"override your config value. Run `unset {setting}` to clear it."
79 )
81 exit_with_success(f"Updated profile {new_profile.name!r}.")
84@config_app.command() 1a
85def validate(): 1a
86 """
87 Read and validate the current profile.
89 Deprecated settings will be automatically converted to new names unless both are
90 set.
91 """
92 profiles = prefect.settings.load_profiles()
93 profile = profiles[prefect.context.get_settings_context().profile.name]
95 profile.validate_settings()
97 prefect.settings.save_profiles(profiles)
98 exit_with_success("Configuration valid!")
101@config_app.command() 1a
102def unset(setting_names: list[str], confirm: bool = typer.Option(False, "--yes", "-y")): 1a
103 """
104 Restore the default value for a setting.
106 Removes the setting from the current profile.
107 """
108 settings_context = prefect.context.get_settings_context()
109 profiles = prefect.settings.load_profiles()
110 profile = profiles[settings_context.profile.name]
111 parsed: set[Setting] = set()
113 for setting_name in setting_names:
114 if setting_name not in VALID_SETTING_NAMES:
115 exit_with_error(f"Unknown setting name {setting_name!r}.")
116 # Cast to settings objects
117 parsed.add(_get_settings_fields(prefect.settings.Settings)[setting_name])
119 for setting in parsed:
120 if setting not in profile.settings:
121 exit_with_error(f"{setting.name!r} is not set in profile {profile.name!r}.")
123 if (
124 not confirm
125 and is_interactive()
126 and not typer.confirm(
127 f"Are you sure you want to unset the following setting(s): {listrepr(setting_names)}?",
128 )
129 ):
130 exit_with_error("Unset aborted.")
132 profiles.update_profile(
133 name=profile.name, settings={setting_name: None for setting_name in parsed}
134 )
136 for setting_name in setting_names:
137 app.console.print(f"Unset {setting_name!r}.")
139 if setting_name in os.environ:
140 app.console.print(
141 f"[yellow]{setting_name!r} is also set by an environment variable. "
142 f"Use `unset {setting_name}` to clear it."
143 )
145 prefect.settings.save_profiles(profiles)
146 exit_with_success(f"Updated profile {profile.name!r}.")
149show_defaults_help = """ 1a
150Toggle display of default settings.
152--show-defaults displays all settings,
153even if they are not changed from the
154default values.
156--hide-defaults displays only settings
157that are changed from default values.
159"""
161show_sources_help = """ 1a
162Toggle display of the source of a value for
163a setting.
165The value for a setting can come from the
166current profile, environment variables, or
167the defaults.
169"""
172@config_app.command() 1a
173def view( 1a
174 show_defaults: bool = typer.Option(
175 False, "--show-defaults/--hide-defaults", help=(show_defaults_help)
176 ),
177 show_sources: bool = typer.Option(
178 True,
179 "--show-sources/--hide-sources",
180 help=(show_sources_help),
181 ),
182 show_secrets: bool = typer.Option(
183 False,
184 "--show-secrets/--hide-secrets",
185 help="Toggle display of secrets setting values.",
186 ),
187):
188 """
189 Display the current settings.
190 """
191 if show_secrets:
192 dump_context = dict(include_secrets=True)
193 else:
194 dump_context = {}
196 context = prefect.context.get_settings_context()
197 current_profile_settings = context.profile.settings
199 if ui_url := prefect.settings.PREFECT_UI_URL.value():
200 app.console.print(
201 f"🚀 you are connected to:\n[green]{ui_url}[/green]", soft_wrap=True
202 )
204 # Display the profile first
205 app.console.print(f"[bold][blue]PREFECT_PROFILE={context.profile.name!r}[/bold]")
207 settings_output: list[str] = []
208 processed_settings: set[str] = set()
210 def _process_setting(
211 setting: prefect.settings.Setting,
212 value: str,
213 source: Literal[
214 "env", "profile", "defaults", ".env file", "prefect.toml", "pyproject.toml"
215 ],
216 ):
217 display_value = "********" if setting.is_secret and not show_secrets else value
218 source_blurb = f" (from {source})" if show_sources else ""
219 settings_output.append(f"{setting.name}='{display_value}'{source_blurb}")
220 processed_settings.add(setting.name)
222 def _collect_defaults(default_values: dict[str, Any], current_path: list[str]):
223 for key, value in default_values.items():
224 if isinstance(value, dict):
225 _collect_defaults(cast(dict[str, Any], value), current_path + [key])
226 else:
227 setting = _get_settings_fields(prefect.settings.Settings)[
228 ".".join(current_path + [key])
229 ]
230 if setting.name in processed_settings:
231 continue
232 _process_setting(setting, value, "defaults")
234 def _process_toml_settings(
235 settings: dict[str, Any],
236 base_path: list[str],
237 source: Literal["prefect.toml", "pyproject.toml"],
238 ):
239 for key, value in settings.items():
240 if isinstance(value, dict):
241 _process_toml_settings(
242 cast(dict[str, Any], value), base_path + [key], source
243 )
244 else:
245 setting = _get_settings_fields(prefect.settings.Settings).get(
246 ".".join(base_path + [key]), NotSet
247 )
248 if setting is NotSet:
249 continue
250 elif (
251 isinstance(setting, Setting) and setting.name in processed_settings
252 ):
253 continue
254 elif isinstance(setting, Setting):
255 _process_setting(setting, value, source)
257 # Process settings from environment variables
258 for setting_name in VALID_SETTING_NAMES:
259 setting = _get_settings_fields(prefect.settings.Settings)[setting_name]
260 if setting.name in processed_settings:
261 continue
262 if (env_value := os.getenv(setting.name)) is None:
263 continue
264 _process_setting(setting, env_value, "env")
266 # Process settings from .env file
267 for key, value in dotenv_values(".env").items():
268 if key in VALID_SETTING_NAMES:
269 setting = _get_settings_fields(prefect.settings.Settings)[key]
270 if setting.name in processed_settings or value is None:
271 continue
272 _process_setting(setting, value, ".env file")
274 # Process settings from prefect.toml
275 if Path("prefect.toml").exists():
276 toml_settings = toml.load(Path("prefect.toml"))
277 _process_toml_settings(toml_settings, base_path=[], source="prefect.toml")
279 # Process settings from pyproject.toml
280 if Path("pyproject.toml").exists():
281 pyproject_settings = toml.load(Path("pyproject.toml"))
282 pyproject_settings = pyproject_settings.get("tool", {}).get("prefect", {})
283 _process_toml_settings(
284 pyproject_settings, base_path=[], source="pyproject.toml"
285 )
287 # Process settings from the current profile
288 for setting, value in current_profile_settings.items():
289 if setting.name not in processed_settings:
290 _process_setting(setting, value, "profile")
292 if show_defaults:
293 _collect_defaults(
294 prefect.settings.Settings().model_dump(context=dump_context),
295 current_path=[],
296 )
298 app.console.print("\n".join(sorted(settings_output)), soft_wrap=True)