Remove legacy settings imports from CLI (#19353)

Author: desertaxle

src/prefect/cli/_types.py

@@ -6,19 +6,22 @@
 import functools
 import sys
 from datetime import datetime
-from typing import Any, Callable, List, Optional
+from typing import TYPE_CHECKING, Any, Callable, List, Optional
 
 import typer
 from rich.console import Console
 from rich.theme import Theme
 
 from prefect._internal.compatibility.deprecated import generate_deprecation_message
 from prefect.cli._utilities import with_cli_exception_handling
-from prefect.settings import PREFECT_CLI_COLORS, Setting
+from prefect.settings import get_current_settings
 from prefect.utilities.asyncutils import is_async_fn
 
+if TYPE_CHECKING:
+    from prefect.settings.legacy import Setting
 
-def SettingsOption(setting: Setting, *args: Any, **kwargs: Any) -> Any:
+
+def SettingsOption(setting: "Setting", *args: Any, **kwargs: Any) -> Any:
     """Custom `typer.Option` factory to load the default value from settings"""
 
     return typer.Option(
@@ -34,7 +37,7 @@ def SettingsOption(setting: Setting, *args: Any, **kwargs: Any) -> Any:
     )
 
 
-def SettingsArgument(setting: Setting, *args: Any, **kwargs: Any) -> Any:
+def SettingsArgument(setting: "Setting", *args: Any, **kwargs: Any) -> Any:
     """Custom `typer.Argument` factory to load the default value from settings"""
 
     # See comments in `SettingsOption`
@@ -91,7 +94,7 @@ def __init__(
         self.console = Console(
             highlight=False,
             theme=Theme({"prompt.choices": "bold blue"}),
-            color_system="auto" if PREFECT_CLI_COLORS else None,
+            color_system="auto" if get_current_settings().cli.colors else None,
         )
 
     def add_typer(
@@ -201,7 +204,7 @@ def sync_fn(*args: Any, **kwargs: Any) -> Any:
     def setup_console(self, soft_wrap: bool, prompt: bool) -> None:
         self.console = Console(
             highlight=False,
-            color_system="auto" if PREFECT_CLI_COLORS else None,
+            color_system="auto" if get_current_settings().cli.colors else None,
             theme=Theme({"prompt.choices": "bold blue"}),
             soft_wrap=not soft_wrap,
             force_interactive=prompt,

src/prefect/cli/_utilities.py

@@ -12,7 +12,7 @@
 from click.exceptions import ClickException
 
 from prefect.exceptions import MissingProfileError
-from prefect.settings import PREFECT_TEST_MODE
+from prefect.settings import get_current_settings
 
 
 def exit_with_error(message: str | Exception, code: int = 1, **kwargs: Any) -> NoReturn:
@@ -47,7 +47,7 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
         except MissingProfileError as exc:
             exit_with_error(exc)
         except Exception:
-            if PREFECT_TEST_MODE.value():
+            if get_current_settings().testing.test_mode:
                 raise  # Reraise exceptions during test mode
             traceback.print_exc()
             exit_with_error("An exception occurred.")

src/prefect/cli/block.py

@@ -28,7 +28,7 @@
     ScriptError,
     exception_traceback,
 )
-from prefect.settings import PREFECT_UI_URL
+from prefect.settings import get_current_settings
 from prefect.utilities.asyncutils import run_sync_in_worker_thread
 from prefect.utilities.importtools import load_script_as_module
 
@@ -244,7 +244,7 @@ async def register(
         "go to the Blocks page in the Prefect UI.\n"
     )
 
-    if ui_url := PREFECT_UI_URL:
+    if ui_url := get_current_settings().ui_url:
         if determine_server_type() == ServerType.CLOUD:
             block_catalog_url = f"{ui_url}/settings/blocks/catalog"
         else:
@@ -349,17 +349,16 @@ async def block_create(
             app.console.print(f"Available block types: {', '.join(slugs)}")
             raise typer.Exit(1)
 
-        if not PREFECT_UI_URL:
+        ui_url = get_current_settings().ui_url
+        if not ui_url:
             exit_with_error(
                 "Prefect must be configured to use a hosted Prefect server or "
                 "Prefect Cloud to display the Prefect UI"
             )
         if determine_server_type() == ServerType.CLOUD:
-            block_link = f"{PREFECT_UI_URL.value()}/settings/blocks/catalog/{block_type.slug}/create"
+            block_link = f"{ui_url}/settings/blocks/catalog/{block_type.slug}/create"
         else:
-            block_link = (
-                f"{PREFECT_UI_URL.value()}/blocks/catalog/{block_type.slug}/create"
-            )
+            block_link = f"{ui_url}/blocks/catalog/{block_type.slug}/create"
         app.console.print(
             f"Create a {block_type_slug} block: {block_link}",
         )

src/prefect/cli/cloud/webhook.py

@@ -14,7 +14,7 @@
 from prefect.cli.root import app, is_interactive
 from prefect.client.cloud import get_cloud_client
 from prefect.exceptions import ObjectNotFound
-from prefect.settings import PREFECT_API_URL
+from prefect.settings import get_current_settings
 
 webhook_app: PrefectTyper = PrefectTyper(
     name="webhook", help="Manage Prefect Cloud Webhooks"
@@ -49,7 +49,7 @@ async def ls():
     confirm_logged_in()
 
     # The /webhooks API lives inside the /accounts/{id}/workspaces/{id} routing tree
-    async with get_cloud_client(host=PREFECT_API_URL.value()) as client:
+    async with get_cloud_client(host=get_current_settings().api.url) as client:
         retrieved_webhooks = await client.request("POST", "/webhooks/filter")
         display_table = _render_webhooks_into_table(retrieved_webhooks)
         app.console.print(display_table)
@@ -63,7 +63,7 @@ async def get(webhook_id: UUID):
     confirm_logged_in()
 
     # The /webhooks API lives inside the /accounts/{id}/workspaces/{id} routing tree
-    async with get_cloud_client(host=PREFECT_API_URL.value()) as client:
+    async with get_cloud_client(host=get_current_settings().api.url) as client:
         webhook = await client.request("GET", f"/webhooks/{webhook_id}")
         display_table = _render_webhooks_into_table([webhook])
         app.console.print(display_table)
@@ -94,7 +94,7 @@ async def create(
     confirm_logged_in()
 
     # The /webhooks API lives inside the /accounts/{id}/workspaces/{id} routing tree
-    async with get_cloud_client(host=PREFECT_API_URL.value()) as client:
+    async with get_cloud_client(host=get_current_settings().api.url) as client:
         response = await client.request(
             "POST",
             "/webhooks/",
@@ -122,7 +122,7 @@ async def rotate(webhook_id: UUID):
         return
 
     # The /webhooks API lives inside the /accounts/{id}/workspaces/{id} routing tree
-    async with get_cloud_client(host=PREFECT_API_URL.value()) as client:
+    async with get_cloud_client(host=get_current_settings().api.url) as client:
         response = await client.request("POST", f"/webhooks/{webhook_id}/rotate")
         app.console.print(f"Successfully rotated webhook URL to {response['slug']}")
 
@@ -138,7 +138,7 @@ async def toggle(
 
     status_lookup = {True: "enabled", False: "disabled"}
 
-    async with get_cloud_client(host=PREFECT_API_URL.value()) as client:
+    async with get_cloud_client(host=get_current_settings().api.url) as client:
         response = await client.request("GET", f"/webhooks/{webhook_id}")
         current_status = response["enabled"]
         new_status = not current_status
@@ -166,7 +166,7 @@ async def update(
     confirm_logged_in()
 
     # The /webhooks API lives inside the /accounts/{id}/workspaces/{id} routing tree
-    async with get_cloud_client(host=PREFECT_API_URL.value()) as client:
+    async with get_cloud_client(host=get_current_settings().api.url) as client:
         response = await client.request("GET", f"/webhooks/{webhook_id}")
         update_payload = {
             "name": webhook_name or response["name"],
@@ -192,7 +192,7 @@ async def delete(webhook_id: UUID):
         exit_with_error("Deletion aborted.")
 
     # The /webhooks API lives inside the /accounts/{id}/workspaces/{id} routing tree
-    async with get_cloud_client(host=PREFECT_API_URL.value()) as client:
+    async with get_cloud_client(host=get_current_settings().api.url) as client:
         try:
             await client.request("DELETE", f"/webhooks/{webhook_id}")
             app.console.print(f"Successfully deleted webhook {webhook_id}")

src/prefect/cli/dashboard.py

@@ -3,7 +3,7 @@
 from prefect.cli._types import PrefectTyper
 from prefect.cli._utilities import exit_with_error, exit_with_success
 from prefect.cli.root import app
-from prefect.settings import PREFECT_UI_URL
+from prefect.settings import get_current_settings
 from prefect.utilities.asyncutils import run_sync_in_worker_thread
 
 dashboard_app: PrefectTyper = PrefectTyper(
@@ -19,7 +19,7 @@ async def open() -> None:
     Open the Prefect UI in the browser.
     """
 
-    if not (ui_url := PREFECT_UI_URL.value()):
+    if not (ui_url := get_current_settings().ui_url):
         exit_with_error(
             "`PREFECT_UI_URL` must be set to the URL of a running Prefect server or Prefect Cloud workspace."
         )

src/prefect/cli/deploy/_commands.py

@@ -8,12 +8,11 @@
 from rich.table import Table
 
 import prefect
-from prefect.cli._types import SettingsOption
 from prefect.cli._utilities import exit_with_error
 from prefect.cli.root import app, is_interactive
 from prefect.client.schemas.objects import ConcurrencyLimitConfig
 from prefect.deployments import initialize_project
-from prefect.settings import PREFECT_DEFAULT_WORK_POOL_NAME
+from prefect.settings import get_current_settings
 
 from ._config import (
     _load_deploy_configs_and_actions,
@@ -178,11 +177,12 @@ async def deploy(
         "--collision-strategy",
         help="Configure the behavior for runs once the concurrency limit is reached. Falls back to `ENQUEUE` if unset.",
     ),
-    work_pool_name: str = SettingsOption(
-        PREFECT_DEFAULT_WORK_POOL_NAME,
+    work_pool_name: str = typer.Option(
+        lambda: get_current_settings().deployments.default_work_pool_name,
         "-p",
         "--pool",
         help="The work pool that will handle this deployment's runs.",
+        show_default="from PREFECT_DEFAULT_WORK_POOL_NAME",
     ),
     work_queue_name: str = typer.Option(
         None,

src/prefect/cli/deploy/_core.py

@@ -25,7 +25,7 @@
 from prefect.deployments.steps.core import run_steps
 from prefect.exceptions import ObjectNotFound
 from prefect.flows import load_flow_from_entrypoint
-from prefect.settings import PREFECT_UI_URL
+from prefect.settings import get_current_settings
 from prefect.utilities.callables import parameter_schema
 from prefect.utilities.collections import get_from_dict
 from prefect.utilities.templating import (
@@ -373,10 +373,10 @@ async def _run_single_deploy(
         style="green",
     )
 
-    if PREFECT_UI_URL:
+    if ui_url := get_current_settings().ui_url:
         message = (
             "\nView Deployment in UI:"
-            f" {PREFECT_UI_URL.value()}/deployments/deployment/{deployment_id}\n"
+            f" {ui_url}/deployments/deployment/{deployment_id}\n"
         )
         app.console.print(message, soft_wrap=True)
 

src/prefect/cli/root.py

@@ -19,10 +19,7 @@
 from prefect.client.base import determine_server_type
 from prefect.client.constants import SERVER_API_VERSION
 from prefect.logging.configuration import setup_logging
-from prefect.settings import (
-    PREFECT_CLI_WRAP_LINES,
-    PREFECT_TEST_MODE,
-)
+from prefect.settings import get_current_settings
 from prefect.types._datetime import parse_datetime
 
 app: PrefectTyper = PrefectTyper(add_completion=True, no_args_is_help=True)
@@ -77,9 +74,9 @@ def main(
             exit(1)
 
     # Configure the output console after loading the profile
-    app.setup_console(soft_wrap=PREFECT_CLI_WRAP_LINES.value(), prompt=prompt)
+    app.setup_console(soft_wrap=get_current_settings().cli.wrap_lines, prompt=prompt)
 
-    if not PREFECT_TEST_MODE:
+    if not get_current_settings().testing.test_mode:
         # When testing, this entrypoint can be called multiple times per process which
         # can cause logging configuration conflicts. Logging is set up in conftest
         # during tests.
@@ -107,7 +104,6 @@ async def version(
 
     from prefect.server.database.dependencies import provide_database_interface
     from prefect.server.utilities.database import get_dialect
-    from prefect.settings import PREFECT_API_DATABASE_CONNECTION_URL
 
     if build_date_str := prefect.__version_info__.get("date", None):
         build_date = parse_datetime(build_date_str).strftime("%a, %b %d, %Y %I:%M %p")
@@ -134,7 +130,8 @@ async def version(
 
     version_info["Pydantic version"] = pydantic_version
 
-    if connection_url := PREFECT_API_DATABASE_CONNECTION_URL.value():
+    if connection_url_setting := get_current_settings().server.database.connection_url:
+        connection_url = connection_url_setting.get_secret_value()
         database = get_dialect(connection_url).name
         version_info["Server"] = {"Database": database}
         if database == "sqlite":

src/prefect/cli/shell.py

@@ -22,7 +22,7 @@
 from prefect.exceptions import FailedRun
 from prefect.logging.loggers import get_run_logger
 from prefect.runner import Runner
-from prefect.settings import PREFECT_UI_URL
+from prefect.settings import get_current_settings
 from prefect.types.entrypoint import EntrypointType
 
 shell_app: PrefectTyper = PrefectTyper(
@@ -236,10 +236,10 @@ async def serve(
         " command:\n[blue]\n\t$ prefect deployment run"
         f" '{runner_deployment.flow_name}/{deployment_name}'\n[/]"
     )
-    if PREFECT_UI_URL:
+    if ui_url := get_current_settings().ui_url:
         help_message += (
             "\nYou can also run your flow via the Prefect UI:"
-            f" [blue]{PREFECT_UI_URL.value()}/deployments/deployment/{deployment_id}[/]\n"
+            f" [blue]{ui_url}/deployments/deployment/{deployment_id}[/]\n"
         )
 
     app.console.print(help_message, soft_wrap=True)

src/prefect/settings/__init__.py

@@ -10,11 +10,8 @@
 for settings, at which point we will not need to use the "after" model_validator.
 """
 
-from prefect.settings.legacy import (
-    Setting,
-    _get_settings_fields,
-    _get_valid_setting_names,
-)
+from typing import TYPE_CHECKING
+
 from prefect.settings.models.root import Settings, canonical_environment_prefix
 
 from prefect.settings.profiles import (
@@ -29,11 +26,18 @@
 from prefect.settings.context import get_current_settings, temporary_settings
 from prefect.settings.constants import DEFAULT_PROFILES_PATH
 
+if TYPE_CHECKING:
+    from prefect.settings.legacy import Setting
 ############################################################################
 # Allow traditional env var access
 
 
-def __getattr__(name: str) -> Setting:
+def __getattr__(name: str) -> "Setting":
+    from prefect.settings.legacy import (
+        _get_settings_fields,
+        _get_valid_setting_names,
+    )
+
     if name in _get_valid_setting_names(Settings):
         return _get_settings_fields(Settings)[name]
     raise AttributeError(f"{name} is not a Prefect setting.")
@@ -42,7 +46,6 @@ def __getattr__(name: str) -> Setting:
 __all__ = [  # noqa: F822
     "Profile",
     "ProfilesCollection",
-    "Setting",
     "Settings",
     "load_current_profile",
     "update_current_profile",