Skip to content

prefect.cli.config

Command line interface for working with profiles

set_(settings)

Change the value for a setting by setting the value in the current profile.

Source code in src/prefect/cli/config.py
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@config_app.command("set")
def set_(settings: List[str]):
    """
    Change the value for a setting by setting the value in the current profile.
    """
    parsed_settings = {}
    for item in settings:
        try:
            setting, value = item.split("=", maxsplit=1)
        except ValueError:
            exit_with_error(
                f"Failed to parse argument {item!r}. Use the format 'VAR=VAL'."
            )

        if setting not in VALID_SETTING_NAMES:
            exit_with_error(f"Unknown setting name {setting!r}.")

        # Guard against changing settings that tweak config locations
        if setting in {"PREFECT_HOME", "PREFECT_PROFILES_PATH"}:
            exit_with_error(
                f"Setting {setting!r} cannot be changed with this command. "
                "Use an environment variable instead."
            )

        parsed_settings[setting] = value

    try:
        new_profile = prefect.settings.update_current_profile(parsed_settings)
    except ProfileSettingsValidationError as exc:
        help_message = ""
        for setting, problem in exc.errors:
            for error in problem.errors():
                help_message += f"[bold red]Validation error(s) for setting[/bold red] [blue]{setting.name}[/blue]\n\n - {error['msg']}\n\n"
        exit_with_error(help_message)

    for setting, value in parsed_settings.items():
        app.console.print(f"Set {setting!r} to {value!r}.")
        if setting in os.environ:
            app.console.print(
                f"[yellow]{setting} is also set by an environment variable which will "
                f"override your config value. Run `unset {setting}` to clear it."
            )

    exit_with_success(f"Updated profile {new_profile.name!r}.")

unset(setting_names, confirm=typer.Option(False, '--yes', '-y'))

Restore the default value for a setting.

Removes the setting from the current profile.

Source code in src/prefect/cli/config.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@config_app.command()
def unset(setting_names: List[str], confirm: bool = typer.Option(False, "--yes", "-y")):
    """
    Restore the default value for a setting.

    Removes the setting from the current profile.
    """
    settings_context = prefect.context.get_settings_context()
    profiles = prefect.settings.load_profiles()
    profile = profiles[settings_context.profile.name]
    parsed = set()

    for setting_name in setting_names:
        if setting_name not in VALID_SETTING_NAMES:
            exit_with_error(f"Unknown setting name {setting_name!r}.")
        # Cast to settings objects
        parsed.add(_get_settings_fields(prefect.settings.Settings)[setting_name])

    for setting in parsed:
        if setting not in profile.settings:
            exit_with_error(f"{setting.name!r} is not set in profile {profile.name!r}.")

    if (
        not confirm
        and is_interactive()
        and not typer.confirm(
            f"Are you sure you want to unset the following setting(s): {listrepr(setting_names)}?",
        )
    ):
        exit_with_error("Unset aborted.")

    profiles.update_profile(
        name=profile.name, settings={setting_name: None for setting_name in parsed}
    )

    for setting_name in setting_names:
        app.console.print(f"Unset {setting_name!r}.")

        if setting_name in os.environ:
            app.console.print(
                f"[yellow]{setting_name!r} is also set by an environment variable. "
                f"Use `unset {setting_name}` to clear it."
            )

    prefect.settings.save_profiles(profiles)
    exit_with_success(f"Updated profile {profile.name!r}.")

validate()

Read and validate the current profile.

Deprecated settings will be automatically converted to new names unless both are set.

Source code in src/prefect/cli/config.py
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@config_app.command()
def validate():
    """
    Read and validate the current profile.

    Deprecated settings will be automatically converted to new names unless both are
    set.
    """
    profiles = prefect.settings.load_profiles()
    profile = profiles[prefect.context.get_settings_context().profile.name]

    profile.validate_settings()

    prefect.settings.save_profiles(profiles)
    exit_with_success("Configuration valid!")

view(show_defaults=typer.Option(False, '--show-defaults/--hide-defaults', help=show_defaults_help), show_sources=typer.Option(True, '--show-sources/--hide-sources', help=show_sources_help), show_secrets=typer.Option(False, '--show-secrets/--hide-secrets', help='Toggle display of secrets setting values.'))

Display the current settings.

Source code in src/prefect/cli/config.py
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
@config_app.command()
def view(
    show_defaults: Optional[bool] = typer.Option(
        False, "--show-defaults/--hide-defaults", help=(show_defaults_help)
    ),
    show_sources: Optional[bool] = typer.Option(
        True,
        "--show-sources/--hide-sources",
        help=(show_sources_help),
    ),
    show_secrets: Optional[bool] = typer.Option(
        False,
        "--show-secrets/--hide-secrets",
        help="Toggle display of secrets setting values.",
    ),
):
    """
    Display the current settings.
    """
    if show_secrets:
        dump_context = dict(include_secrets=True)
    else:
        dump_context = {}

    context = prefect.context.get_settings_context()
    current_profile_settings = context.profile.settings

    if ui_url := prefect.settings.PREFECT_UI_URL.value():
        app.console.print(
            f"🚀 you are connected to:\n[green]{ui_url}[/green]", soft_wrap=True
        )

    # Display the profile first
    app.console.print(f"[bold][blue]PREFECT_PROFILE={context.profile.name!r}[/bold]")

    settings_output = []
    processed_settings = set()

    def _process_setting(
        setting: prefect.settings.Setting,
        value: str,
        source: Literal[
            "env", "profile", "defaults", ".env file", "prefect.toml", "pyproject.toml"
        ],
    ):
        display_value = "********" if setting.is_secret and not show_secrets else value
        source_blurb = f" (from {source})" if show_sources else ""
        settings_output.append(f"{setting.name}='{display_value}'{source_blurb}")
        processed_settings.add(setting.name)

    def _collect_defaults(default_values: Dict[str, Any], current_path: List[str]):
        for key, value in default_values.items():
            if isinstance(value, dict):
                _collect_defaults(value, current_path + [key])
            else:
                setting = _get_settings_fields(prefect.settings.Settings)[
                    ".".join(current_path + [key])
                ]
                if setting.name in processed_settings:
                    continue
                _process_setting(setting, value, "defaults")

    def _process_toml_settings(
        settings: Dict[str, Any],
        base_path: List[str],
        source: Literal["prefect.toml", "pyproject.toml"],
    ):
        for key, value in settings.items():
            if isinstance(value, dict):
                _process_toml_settings(value, base_path + [key], source)
            else:
                setting = _get_settings_fields(prefect.settings.Settings).get(
                    ".".join(base_path + [key]), NotSet
                )
                if setting is NotSet or setting.name in processed_settings:
                    continue
                _process_setting(setting, value, source)

    # Process settings from environment variables
    for setting_name in VALID_SETTING_NAMES:
        setting = _get_settings_fields(prefect.settings.Settings)[setting_name]
        if setting.name in processed_settings:
            continue
        if (env_value := os.getenv(setting.name)) is None:
            continue
        _process_setting(setting, env_value, "env")

    # Process settings from .env file
    for key, value in dotenv_values().items():
        if key in VALID_SETTING_NAMES:
            setting = _get_settings_fields(prefect.settings.Settings)[key]
            if setting.name in processed_settings or value is None:
                continue
            _process_setting(setting, value, ".env file")

    # Process settings from prefect.toml
    if Path("prefect.toml").exists():
        toml_settings = toml.load(Path("prefect.toml"))
        _process_toml_settings(toml_settings, base_path=[], source="prefect.toml")

    # Process settings from pyproject.toml
    if Path("pyproject.toml").exists():
        pyproject_settings = toml.load(Path("pyproject.toml"))
        pyproject_settings = pyproject_settings.get("tool", {}).get("prefect", {})
        _process_toml_settings(
            pyproject_settings, base_path=[], source="pyproject.toml"
        )

    # Process settings from the current profile
    for setting, value in current_profile_settings.items():
        if setting.name not in processed_settings:
            _process_setting(setting, value, "profile")

    if show_defaults:
        _collect_defaults(
            prefect.settings.Settings().model_dump(context=dump_context),
            current_path=[],
        )

    app.console.print("\n".join(sorted(settings_output)), soft_wrap=True)