add prefect profiles populate-defaults CLI command (#14749)

src/prefect/cli/profile.py

@@ -3,6 +3,7 @@
 """
 
 import os
+import shutil
 import textwrap
 from typing import Optional
 
@@ -25,6 +26,10 @@
 profile_app = PrefectTyper(name="profile", help="Select and manage Prefect profiles.")
 app.add_typer(profile_app, aliases=["profiles"])
 
+_OLD_MINIMAL_DEFAULT_PROFILE_CONTENT: str = """active = "default"
+
+[profiles.default]"""
+
 
 @profile_app.command()
 def ls():
@@ -250,6 +255,49 @@ def inspect(
         app.console.print(f"{setting.name}='{value}'")
 
 
+@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._read_profiles_from(
+        prefect.settings.DEFAULT_PROFILES_PATH
+    )
+
+    if user_path.exists():
+        user_content = user_path.read_text()
+        if user_content == prefect.settings.DEFAULT_PROFILES_PATH.read_text():
+            app.console.print(
+                "Default profiles already populated. [green]No action required[/green]."
+            )
+            return
+
+        if user_content != _OLD_MINIMAL_DEFAULT_PROFILE_CONTENT:
+            backup_path = user_path.with_suffix(".toml.bak")
+            if typer.confirm(f"Back up existing profiles to {backup_path}?"):
+                shutil.copy(user_path, backup_path)
+                app.console.print(f"Profiles backed up to {backup_path}")
+
+        user_profiles = prefect.settings._read_profiles_from(user_path)
+
+        # Merge profiles, keeping existing user profiles unchanged
+        for name, profile in default_profiles.items():
+            if name not in user_profiles:
+                user_profiles.add_profile(profile)
+                app.console.print(f"Added default profile: [blue]{name}[/blue]")
+    else:
+        user_profiles = default_profiles
+
+    if not typer.confirm(f"Update profiles at {user_path}?"):
+        app.console.print("Operation cancelled.")
+        return
+
+    prefect.settings._write_profiles_to(user_path, user_profiles)
+    app.console.print(f"Profiles updated in [green]{user_path}[/green]")
+    app.console.print(
+        "Use with [green]prefect profile use[/green] [red][PROFILE-NAME][/red]"
+    )
+
+
 class ConnectionStatus(AutoEnum):
     CLOUD_CONNECTED = AutoEnum.auto()
     CLOUD_ERROR = AutoEnum.auto()

src/prefect/profiles.toml

@@ -1,3 +1,20 @@
+# This is a template for profile configuration for Prefect.
+# You can modify these profiles or create new ones to suit their needs.
+
 active = "default"
 
-[profiles.default]
+[profiles.default]
+
+[profiles.ephemeral]
+PREFECT_API_DATABASE_CONNECTION_URL = "sqlite+aiosqlite:///prefect.db"
+
+[profiles.local]
+# You will need to set these values appropriately
+PREFECT_API_URL = "http://127.0.0.1:4200/api"
+
+
+[profiles.cloud]
+# This is a placeholder for Prefect Cloud configuration
+# Run "prefect cloud login" to set up your Prefect Cloud profile
+# PREFECT_API_URL = "https://api.prefect.cloud/api/accounts/[ACCOUNT-ID]/workspaces/[WORKSPACE-ID]"
+# PREFECT_API_KEY = "pnu_rest-of-your-api-key-here"

src/prefect/settings.py

@@ -2145,10 +2145,10 @@ def load_current_profile():
     This will _not_ include settings from the current settings context. Only settings
     that have been persisted to the profiles file will be saved.
     """
-    from prefect.context import SettingsContext
+    import prefect.context
 
     profiles = load_profiles()
-    context = SettingsContext.get()
+    context = prefect.context.get_settings_context()
 
     if context:
         profiles.set_active(context.profile.name)

tests/cli/test_profile.py

@@ -1,6 +1,7 @@
 from uuid import uuid4
 
 import pytest
+import readchar
 import respx
 from httpx import Response
 
@@ -12,6 +13,7 @@
     PREFECT_PROFILES_PATH,
     Profile,
     ProfilesCollection,
+    _read_profiles_from,
     load_profiles,
     save_profiles,
     temporary_settings,
@@ -581,3 +583,66 @@ def test_inspect_profile_without_settings():
             Profile 'foo' is empty.
             """,
     )
+
+
+def test_populate_defaults(tmp_path, monkeypatch):
+    # Set up a temporary profiles path
+    temp_profiles_path = tmp_path / "profiles.toml"
+    monkeypatch.setattr(PREFECT_PROFILES_PATH, "value", lambda: temp_profiles_path)
+
+    default_profiles = _read_profiles_from(DEFAULT_PROFILES_PATH)
+
+    assert not temp_profiles_path.exists()
+
+    invoke_and_assert(
+        ["profile", "populate-defaults"],
+        user_input="y",
+        expected_output_contains=[
+            f"Profiles updated in {temp_profiles_path}",
+            "Use with prefect profile use [PROFILE-NAME]",
+        ],
+    )
+
+    assert temp_profiles_path.exists()
+
+    populated_profiles = load_profiles()
+
+    assert populated_profiles.names == default_profiles.names
+    assert populated_profiles.active_name == default_profiles.active_name
+
+    assert {"local", "ephemeral", "cloud", "default"} == set(populated_profiles.names)
+
+    for name in default_profiles.names:
+        assert populated_profiles[name].settings == default_profiles[name].settings
+
+
+def test_populate_defaults_with_existing_profiles(tmp_path, monkeypatch):
+    temp_profiles_path = tmp_path / "profiles.toml"
+    monkeypatch.setattr(PREFECT_PROFILES_PATH, "value", lambda: temp_profiles_path)
+
+    existing_profiles = ProfilesCollection(
+        profiles=[Profile(name="existing", settings={PREFECT_API_KEY: "test_key"})],
+        active="existing",
+    )
+    save_profiles(existing_profiles)
+
+    invoke_and_assert(
+        ["profile", "populate-defaults"],
+        user_input=(
+            "y" + readchar.key.ENTER + "y" + readchar.key.ENTER
+        ),  # Confirm overwrite and backup
+        expected_output_contains=[
+            "Update profiles at",
+            f"Profiles backed up to {temp_profiles_path}.bak",
+            f"Profiles updated in {temp_profiles_path}",
+        ],
+    )
+
+    new_profiles = load_profiles()
+    assert {"local", "ephemeral", "cloud", "default", "existing"} == set(
+        new_profiles.names
+    )
+
+    backup_profiles = _read_profiles_from(temp_profiles_path.with_suffix(".toml.bak"))
+    assert "existing" in backup_profiles.names
+    assert backup_profiles["existing"].settings == {PREFECT_API_KEY: "test_key"}

tests/test_context.py

@@ -299,7 +299,7 @@ def test_root_settings_context_default(self, monkeypatch):
             Profile(name="default", settings={}, source=DEFAULT_PROFILES_PATH),
             override_environment_variables=False,
         )
-        use_profile().__enter__.assert_called_once_with()
+        use_profile().__enter__.assert_called_once()
         assert result is not None
 
     @pytest.mark.parametrize(

tests/test_settings.py

@@ -569,31 +569,22 @@ def test_load_profiles_empty_file(self, temporary_profiles_path):
 
     def test_load_profiles_with_default(self, temporary_profiles_path):
         temporary_profiles_path.write_text(
-            textwrap.dedent(
-                """
-                [profiles.default]
-                PREFECT_API_KEY = "foo"
+            """
+            [profiles.default]
+            PREFECT_API_KEY = "foo"
 
-                [profiles.bar]
-                PREFECT_API_KEY = "bar"
-                """
-            )
-        )
-        assert load_profiles() == ProfilesCollection(
-            profiles=[
-                Profile(
-                    name="default",
-                    settings={PREFECT_API_KEY: "foo"},
-                    source=temporary_profiles_path,
-                ),
-                Profile(
-                    name="bar",
-                    settings={PREFECT_API_KEY: "bar"},
-                    source=temporary_profiles_path,
-                ),
-            ],
-            active="default",
+            [profiles.bar]
+            PREFECT_API_KEY = "bar"
+            """
         )
+        profiles = load_profiles()
+        expected = {
+            "default": {PREFECT_API_KEY: "foo"},
+            "bar": {PREFECT_API_KEY: "bar"},
+        }
+        for name, settings in expected.items():
+            assert profiles[name].settings == settings
+            assert profiles[name].source == temporary_profiles_path
 
     def test_load_profile_default(self):
         assert load_profile("default") == Profile(