Automatically run a server in a subprocess if no API URL is specified

.github/workflows/benchmarks.yaml

@@ -105,8 +105,8 @@ jobs:
           # characters with an underscore
           sanitized_uniquename="${uniquename//[^a-zA-Z0-9_\-]/_}"
 
-          PREFECT_API_URL="http://127.0.0.1:4200/api"
-          python benches \
+          PREFECT_API_URL="http://127.0.0.1:4200/api" \
+          python -m benches \
           --ignore=benches/bench_import.py \
           --timeout=180 \
           --benchmark-save="${sanitized_uniquename}" \

benches/bench_flows.py

@@ -55,7 +55,10 @@ def benchmark_flow():
         for _ in range(num_tasks):
             test_task()
 
-    benchmark(benchmark_flow)
+    if num_tasks > 100:
+        benchmark.pedantic(benchmark_flow)
+    else:
+        benchmark(benchmark_flow)
 
 
 @pytest.mark.parametrize("num_tasks", [10, 50, 100, 250])
@@ -68,7 +71,10 @@ async def benchmark_flow():
             for _ in range(num_tasks):
                 tg.start_soon(test_task)
 
-    benchmark(anyio.run, benchmark_flow)
+    if num_tasks > 100:
+        benchmark.pedantic(anyio.run, (benchmark_flow,))
+    else:
+        benchmark(anyio.run, benchmark_flow)
 
 
 @pytest.mark.parametrize("num_flows", [5, 10, 20])

docs/3.0rc/api-ref/rest-api/server/schema.json

@@ -22271,6 +22271,16 @@
                         "title": "Prefect Server Csrf Token Expiration",
                         "default": "PT1H"
                     },
+                    "PREFECT_SERVER_ALLOW_EPHEMERAL_MODE": {
+                        "type": "boolean",
+                        "title": "Prefect Server Allow Ephemeral Mode",
+                        "default": false
+                    },
+                    "PREFECT_SERVER_EPHEMERAL_START_UP_WAIT_SECONDS": {
+                        "type": "integer",
+                        "title": "Prefect Server Ephemeral Start Up Wait Seconds",
+                        "default": 10
+                    },
                     "PREFECT_UI_ENABLED": {
                         "type": "boolean",
                         "title": "Prefect Ui Enabled",

src/prefect/blocks/core.py

@@ -902,7 +902,9 @@ class Custom(Block):
             loaded_block.save("my-custom-message", overwrite=True)
             ```
         """
-        block_document, block_document_name = await cls._get_block_document(name)
+        block_document, block_document_name = await cls._get_block_document(
+            name, client=client
+        )
 
         return cls._load_from_block_document(block_document, validate=validate)
 

src/prefect/cli/profile.py

@@ -18,6 +18,7 @@
 from prefect.cli._utilities import exit_with_error, exit_with_success
 from prefect.cli.cloud import CloudUnauthorizedError, get_cloud_client
 from prefect.cli.root import app, is_interactive
+from prefect.client.base import determine_server_type
 from prefect.client.orchestration import ServerType, get_client
 from prefect.context import use_profile
 from prefect.exceptions import ObjectNotFound
@@ -122,11 +123,11 @@ async def use(name: str):
             exit_with_error,
             f"Error authenticating with Prefect Cloud using profile {name!r}",
         ),
-        ConnectionStatus.ORION_CONNECTED: (
+        ConnectionStatus.SERVER_CONNECTED: (
             exit_with_success,
             f"Connected to Prefect server using profile {name!r}",
         ),
-        ConnectionStatus.ORION_ERROR: (
+        ConnectionStatus.SERVER_ERROR: (
             exit_with_error,
             f"Error connecting to Prefect server using profile {name!r}",
         ),
@@ -137,6 +138,13 @@ async def use(name: str):
                 " 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",
@@ -302,8 +310,9 @@ class ConnectionStatus(AutoEnum):
     CLOUD_CONNECTED = AutoEnum.auto()
     CLOUD_ERROR = AutoEnum.auto()
     CLOUD_UNAUTHORIZED = AutoEnum.auto()
-    ORION_CONNECTED = AutoEnum.auto()
-    ORION_ERROR = AutoEnum.auto()
+    SERVER_CONNECTED = AutoEnum.auto()
+    SERVER_ERROR = AutoEnum.auto()
+    UNCONFIGURED = AutoEnum.auto()
     EPHEMERAL = AutoEnum.auto()
     INVALID_API = AutoEnum.auto()
 
@@ -327,35 +336,40 @@ async def check_orion_connection():
         try:
             # inform the user if Prefect API endpoints exist, but there are
             # connection issues
+            server_type = determine_server_type()
+            if server_type == ServerType.EPHEMERAL:
+                return ConnectionStatus.EPHEMERAL
+            elif server_type == ServerType.UNCONFIGURED:
+                return ConnectionStatus.UNCONFIGURED
             client = get_client(httpx_settings=httpx_settings)
             async with client:
                 connect_error = await client.api_healthcheck()
             if connect_error is not None:
-                return ConnectionStatus.ORION_ERROR
-            elif client.server_type == ServerType.EPHEMERAL:
-                # if the client is using an ephemeral Prefect app, inform the user
-                return ConnectionStatus.EPHEMERAL
+                return ConnectionStatus.SERVER_ERROR
             else:
-                return ConnectionStatus.ORION_CONNECTED
+                return ConnectionStatus.SERVER_CONNECTED
         except Exception:
-            return ConnectionStatus.ORION_ERROR
+            return ConnectionStatus.SERVER_ERROR
     except httpx.HTTPStatusError:
         return ConnectionStatus.CLOUD_ERROR
     except TypeError:
         # if no Prefect API URL has been set, httpx will throw a TypeError
         try:
             # try to connect with the client anyway, it will likely use an
             # ephemeral Prefect instance
+            server_type = determine_server_type()
+            if server_type == ServerType.EPHEMERAL:
+                return ConnectionStatus.EPHEMERAL
+            elif server_type == ServerType.UNCONFIGURED:
+                return ConnectionStatus.UNCONFIGURED
             client = get_client(httpx_settings=httpx_settings)
             async with client:
                 connect_error = await client.api_healthcheck()
             if connect_error is not None:
-                return ConnectionStatus.ORION_ERROR
-            elif client.server_type == ServerType.EPHEMERAL:
-                return ConnectionStatus.EPHEMERAL
+                return ConnectionStatus.SERVER_ERROR
             else:
-                return ConnectionStatus.ORION_CONNECTED
+                return ConnectionStatus.SERVER_CONNECTED
         except Exception:
-            return ConnectionStatus.ORION_ERROR
+            return ConnectionStatus.SERVER_ERROR
     except (httpx.ConnectError, httpx.UnsupportedProtocol):
         return ConnectionStatus.INVALID_API

src/prefect/cli/root.py

@@ -17,6 +17,7 @@
 import prefect.settings
 from prefect.cli._types import PrefectTyper, SettingsOption
 from prefect.cli._utilities import with_cli_exception_handling
+from prefect.client.base import determine_server_type
 from prefect.client.constants import SERVER_API_VERSION
 from prefect.client.orchestration import ServerType
 from prefect.logging.configuration import setup_logging
@@ -117,16 +118,7 @@ async def version(
         "OS/Arch": f"{sys.platform}/{platform.machine()}",
         "Profile": prefect.context.get_settings_context().profile.name,
     }
-
-    server_type: str
-
-    try:
-        # We do not context manage the client because when using an ephemeral app we do not
-        # want to create the database or run migrations
-        client = prefect.get_client()
-        server_type = client.server_type.value
-    except Exception:
-        server_type = "<client error>"
+    server_type = determine_server_type()
 
     version_info["Server type"] = server_type.lower()
 

src/prefect/client/base.py

@@ -35,10 +35,14 @@
 from prefect.exceptions import PrefectHTTPStatusError
 from prefect.logging import get_logger
 from prefect.settings import (
+    PREFECT_API_URL,
     PREFECT_CLIENT_MAX_RETRIES,
     PREFECT_CLIENT_RETRY_EXTRA_CODES,
     PREFECT_CLIENT_RETRY_JITTER_FACTOR,
+    PREFECT_CLOUD_API_URL,
+    PREFECT_SERVER_ALLOW_EPHEMERAL_MODE,
 )
+from prefect.utilities.collections import AutoEnum
 from prefect.utilities.math import bounded_poisson_interval, clamped_poisson_interval
 
 # Datastores for lifespan management, keys should be a tuple of thread and app
@@ -637,3 +641,33 @@ def __init__(
         )
 
     pass
+
+
+class ServerType(AutoEnum):
+    EPHEMERAL = AutoEnum.auto()
+    SERVER = AutoEnum.auto()
+    CLOUD = AutoEnum.auto()
+    UNCONFIGURED = AutoEnum.auto()
+
+
+def determine_server_type() -> ServerType:
+    """
+    Determine the server type based on the current settings.
+
+    Returns:
+        - `ServerType.EPHEMERAL` if the ephemeral server is enabled
+        - `ServerType.SERVER` if a API URL is configured and it is not a cloud URL
+        - `ServerType.CLOUD` if an API URL is configured and it is a cloud URL
+        - `ServerType.UNCONFIGURED` if no API URL is configured and ephemeral mode is
+            not enabled
+    """
+    api_url = PREFECT_API_URL.value()
+    if api_url is None:
+        if PREFECT_SERVER_ALLOW_EPHEMERAL_MODE.value():
+            return ServerType.EPHEMERAL
+        else:
+            return ServerType.UNCONFIGURED
+    if api_url.startswith(PREFECT_CLOUD_API_URL.value()):
+        return ServerType.CLOUD
+    else:
+        return ServerType.SERVER