Skip to content

prefect.cli.profile

Command line interface for working with profiles.

create(name, from_name=typer.Option(None, '--from', help='Copy an existing profile.'))

Create a new profile.

Source code in src/prefect/cli/profile.py
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@profile_app.command()
def create(
    name: str,
    from_name: str = typer.Option(None, "--from", help="Copy an existing profile."),
):
    """
    Create a new profile.
    """

    profiles = prefect.settings.load_profiles()
    if name in profiles:
        app.console.print(
            textwrap.dedent(
                f"""
                [red]Profile {name!r} already exists.[/red]
                To create a new profile, remove the existing profile first:

                    prefect profile delete {name!r}
                """
            ).strip()
        )
        raise typer.Exit(1)

    if from_name:
        if from_name not in profiles:
            exit_with_error(f"Profile {from_name!r} not found.")

        # Create a copy of the profile with a new name and add to the collection
        profiles.add_profile(profiles[from_name].model_copy(update={"name": name}))
    else:
        profiles.add_profile(prefect.settings.Profile(name=name, settings={}))

    prefect.settings.save_profiles(profiles)

    app.console.print(
        textwrap.dedent(
            f"""
            Created profile with properties:
                name - {name!r}
                from name - {from_name or None}

            Use created profile for future, subsequent commands:
                prefect profile use {name!r}

            Use created profile temporarily for a single command:
                prefect -p {name!r} config view
            """
        )
    )

delete(name)

Delete the given profile.

Source code in src/prefect/cli/profile.py
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@profile_app.command()
def delete(name: str):
    """
    Delete the given profile.
    """
    profiles = prefect.settings.load_profiles()
    if name not in profiles:
        exit_with_error(f"Profile {name!r} not found.")

    current_profile = prefect.context.get_settings_context().profile
    if current_profile.name == name:
        exit_with_error(
            f"Profile {name!r} is the active profile. You must switch profiles before"
            " it can be deleted."
        )
    if is_interactive() and not typer.confirm(
        (f"Are you sure you want to delete profile with name {name!r}?"),
        default=False,
    ):
        exit_with_error("Deletion aborted.")
    profiles.remove_profile(name)

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

inspect(name=typer.Argument(None, help='Name of profile to inspect; defaults to active profile.'))

Display settings from a given profile; defaults to active.

Source code in src/prefect/cli/profile.py
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
@profile_app.command()
def inspect(
    name: Optional[str] = typer.Argument(
        None, help="Name of profile to inspect; defaults to active profile."
    ),
):
    """
    Display settings from a given profile; defaults to active.
    """
    profiles = prefect.settings.load_profiles()
    if name is None:
        current_profile = prefect.context.get_settings_context().profile
        if not current_profile:
            exit_with_error("No active profile set - please provide a name to inspect.")
        name = current_profile.name
        print(f"No name provided, defaulting to {name!r}")
    if name not in profiles:
        exit_with_error(f"Profile {name!r} not found.")

    if not profiles[name].settings:
        # TODO: Consider instructing on how to add settings.
        print(f"Profile {name!r} is empty.")

    for setting, value in profiles[name].settings.items():
        app.console.print(f"{setting.name}='{value}'")

ls()

List profile names.

Source code in src/prefect/cli/profile.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@profile_app.command()
def ls():
    """
    List profile names.
    """
    profiles = prefect.settings.load_profiles(include_defaults=False)
    current_profile = prefect.context.get_settings_context().profile
    current_name = current_profile.name if current_profile is not None else None

    table = Table(caption="* active profile")
    table.add_column(
        "[#024dfd]Available Profiles:", justify="right", style="#8ea0ae", no_wrap=True
    )

    for name in profiles:
        if name == current_name:
            table.add_row(f"[green]  * {name}[/green]")
        else:
            table.add_row(f"  {name}")
    app.console.print(table)

populate_defaults()

Populate the profiles configuration with default base profiles, preserving existing user profiles.

Source code in src/prefect/cli/profile.py
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
@profile_app.command()
def populate_defaults():
    """Populate the profiles configuration with default base profiles, preserving existing user profiles."""
    user_path = prefect.settings.PREFECT_PROFILES_PATH.value()
    default_profiles = prefect.settings.profiles._read_profiles_from(
        prefect.settings.DEFAULT_PROFILES_PATH
    )

    if user_path.exists():
        user_profiles = prefect.settings.profiles._read_profiles_from(user_path)

        if not show_profile_changes(user_profiles, default_profiles):
            return

        # Backup prompt
        if typer.confirm(f"\nBack up existing profiles to {user_path}.bak?"):
            shutil.copy(user_path, f"{user_path}.bak")
            app.console.print(f"Profiles backed up to {user_path}.bak")
    else:
        user_profiles = ProfilesCollection([])
        app.console.print(
            "\n[bold]Creating new profiles file with default profiles.[/bold]"
        )
        show_profile_changes(user_profiles, default_profiles)

    # Update prompt
    if not typer.confirm(f"\nUpdate profiles at {user_path}?"):
        app.console.print("Operation cancelled.")
        return

    for name, profile in default_profiles.items():
        if name not in user_profiles:
            user_profiles.add_profile(profile)

    prefect.settings.profiles._write_profiles_to(user_path, user_profiles)
    app.console.print(f"\nProfiles updated in [green]{user_path}[/green]")
    app.console.print(
        "\nUse with [green]prefect profile use[/green] [blue][PROFILE-NAME][/blue]"
    )
    app.console.print("\nAvailable profiles:")
    for name in user_profiles.names:
        app.console.print(f"  - {name}")

rename(name, new_name)

Change the name of a profile.

Source code in src/prefect/cli/profile.py
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
@profile_app.command()
def rename(name: str, new_name: str):
    """
    Change the name of a profile.
    """
    profiles = prefect.settings.load_profiles()
    if name not in profiles:
        exit_with_error(f"Profile {name!r} not found.")

    if new_name in profiles:
        exit_with_error(f"Profile {new_name!r} already exists.")

    profiles.add_profile(profiles[name].model_copy(update={"name": new_name}))
    profiles.remove_profile(name)

    # If the active profile was renamed switch the active profile to the new name.
    prefect.context.get_settings_context().profile
    if profiles.active_name == name:
        profiles.set_active(new_name)
    if os.environ.get("PREFECT_PROFILE") == name:
        app.console.print(
            f"You have set your current profile to {name!r} with the "
            "PREFECT_PROFILE environment variable. You must update this variable to "
            f"{new_name!r} to continue using the profile."
        )

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

use(name) async

Set the given profile to active.

Source code in src/prefect/cli/profile.py
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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@profile_app.command()
async def use(name: str):
    """
    Set the given profile to active.
    """
    status_messages = {
        ConnectionStatus.CLOUD_CONNECTED: (
            exit_with_success,
            f"Connected to Prefect Cloud using profile {name!r}",
        ),
        ConnectionStatus.CLOUD_ERROR: (
            exit_with_error,
            f"Error connecting to Prefect Cloud using profile {name!r}",
        ),
        ConnectionStatus.CLOUD_UNAUTHORIZED: (
            exit_with_error,
            f"Error authenticating with Prefect Cloud using profile {name!r}",
        ),
        ConnectionStatus.SERVER_CONNECTED: (
            exit_with_success,
            f"Connected to Prefect server using profile {name!r}",
        ),
        ConnectionStatus.SERVER_ERROR: (
            exit_with_error,
            f"Error connecting to Prefect server using profile {name!r}",
        ),
        ConnectionStatus.EPHEMERAL: (
            exit_with_success,
            (
                f"No Prefect server specified using profile {name!r} - the API will run"
                " in ephemeral mode."
            ),
        ),
        ConnectionStatus.UNCONFIGURED: (
            exit_with_error,
            (
                f"Prefect server URL not configured using profile {name!r} - please"
                " configure the server URL or enable ephemeral mode."
            ),
        ),
        ConnectionStatus.INVALID_API: (
            exit_with_error,
            "Error connecting to Prefect API URL",
        ),
    }

    profiles = prefect.settings.load_profiles()
    if name not in profiles.names:
        exit_with_error(f"Profile {name!r} not found.")

    profiles.set_active(name)
    prefect.settings.save_profiles(profiles)

    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        transient=False,
    ) as progress:
        progress.add_task(
            description="Checking API connectivity...",
            total=None,
        )

        with use_profile(name, include_current_context=False):
            connection_status = await check_server_connection()

        exit_method, msg = status_messages[connection_status]

    exit_method(msg)