Experiment with deferring CLI imports

src/prefect/__init__.py

@@ -8,6 +8,8 @@
 import warnings
 import sys
 
+print("HI THERE")
+
 __version_info__ = _version.get_versions()
 __version__ = __version_info__["version"]
 

src/prefect/cli/agent.py

@@ -7,24 +7,19 @@
 from typing import List, Optional
 from uuid import UUID
 
-import anyio
 import typer
 
-import prefect
-from prefect.agent import PrefectAgent
 from prefect.cli._types import PrefectTyper, SettingsOption
-from prefect.cli._utilities import exit_with_error
 from prefect.cli.root import app
-from prefect.client import get_client
-from prefect.client.schemas.filters import WorkQueueFilter, WorkQueueFilterName
-from prefect.exceptions import ObjectNotFound
 from prefect.settings import (
     PREFECT_AGENT_PREFETCH_SECONDS,
     PREFECT_AGENT_QUERY_INTERVAL,
     PREFECT_API_URL,
 )
-from prefect.utilities.processutils import setup_signal_handlers_agent
-from prefect.utilities.services import critical_service_loop
+from prefect.utilities.importtools import lazy_import
+
+anyio = lazy_import("anyio")
+prefect = lazy_import("prefect")
 
 agent_app = PrefectTyper(
     name="agent",
@@ -36,7 +31,6 @@
 )
 app.add_typer(agent_app)
 
-
 ascii_name = r"""
   ___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
  | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
@@ -102,6 +96,13 @@ async def start(
     """
     Start an agent process to poll one or more work queues for flow runs.
     """
+    from prefect.agent import PrefectAgent
+    from prefect.cli._utilities import exit_with_error
+    from prefect.client import get_client
+    from prefect.exceptions import ObjectNotFound
+    from prefect.utilities.processutils import setup_signal_handlers_agent
+    from prefect.utilities.services import critical_service_loop
+
     work_queues = work_queues or []
 
     if work_queue is not None:
@@ -261,6 +262,10 @@ async def _check_work_queues_paused(
     Returns:
         - bool: True if work queues are paused, False otherwise
     """
+    from prefect.client import get_client
+    from prefect.client.schemas.filters import WorkQueueFilter, WorkQueueFilterName
+    from prefect.exceptions import ObjectNotFound
+
     work_queues_list = work_queues or ["default"]
     try:
         work_queues_filter = WorkQueueFilter(

src/prefect/cli/artifact.py

@@ -1,17 +1,12 @@
 from typing import Optional
 
-import pendulum
 import typer
-from rich.pretty import Pretty
-from rich.table import Table
 
-from prefect import get_client
 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.client.schemas.filters import ArtifactFilter, ArtifactFilterKey
-from prefect.client.schemas.sorting import ArtifactCollectionSort, ArtifactSort
-from prefect.exceptions import ObjectNotFound
+from prefect.utilities.importtools import lazy_import
+
+rich = lazy_import("rich")
 
 artifact_app = PrefectTyper(
     name="artifact", help="Commands for starting and interacting with artifacts."
@@ -36,7 +31,12 @@ async def list_artifacts(
     """
     List artifacts.
     """
-    table = Table(
+    import pendulum
+
+    from prefect import get_client
+    from prefect.client.schemas.sorting import ArtifactCollectionSort, ArtifactSort
+
+    table = rich.Table(
         title="Artifacts",
         caption="List Artifacts using `prefect artifact ls`",
         show_header=True,
@@ -123,6 +123,10 @@ async def inspect(
             }
     ]
     """
+    from prefect import get_client
+    from prefect.cli._utilities import exit_with_error
+    from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterKey
+    from prefect.client.schemas.sorting import ArtifactSort
 
     async with get_client() as client:
         artifacts = await client.read_artifacts(
@@ -135,7 +139,7 @@ async def inspect(
 
         artifacts = [a.dict(json_compatible=True) for a in artifacts]
 
-        app.console.print(Pretty(artifacts))
+        app.console.print(rich.Pretty(artifacts))
 
 
 @artifact_app.command("delete")
@@ -156,6 +160,11 @@ async def delete(
     Examples:
         $ prefect artifact delete "my-artifact"
     """
+    from prefect import get_client
+    from prefect.cli._utilities import exit_with_error, exit_with_success
+    from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterKey
+    from prefect.exceptions import ObjectNotFound
+
     if key and artifact_id:
         exit_with_error("Please provide either a key or an artifact_id but not both.")
 

src/prefect/cli/block.py

@@ -1,30 +1,29 @@
 """
 Command line interface for working with blocks.
 """
+
 import inspect
-from importlib import import_module
 from pathlib import Path
 from types import ModuleType
 from typing import List, Optional, Type
 
 import typer
-from rich.table import Table
+from typing_extensions import TYPE_CHECKING
 
-from prefect.blocks.core import Block, InvalidBlockRegistration
 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.client import get_client
 from prefect.exceptions import (
     ObjectNotFound,
     PrefectHTTPStatusError,
-    ProtectedBlockError,
     ScriptError,
     exception_traceback,
 )
-from prefect.settings import PREFECT_UI_URL
-from prefect.utilities.asyncutils import run_sync_in_worker_thread
-from prefect.utilities.importtools import load_script_as_module
+from prefect.utilities.importtools import lazy_import
+
+if TYPE_CHECKING:
+    from prefect.blocks.core import Block
+
+rich = lazy_import("rich")
 
 blocks_app = PrefectTyper(name="block", help="Commands for working with blocks.")
 blocktypes_app = PrefectTyper(
@@ -36,7 +35,7 @@
 
 def display_block(block_document):
     block_slug = f"{block_document.block_type.slug}/{block_document.name}"
-    block_table = Table(
+    block_table = rich.Table(
         title=block_slug, show_header=False, show_footer=False, expand=True
     )
     block_table.add_column(style="italic cyan")
@@ -51,7 +50,7 @@ def display_block(block_document):
 
 
 def display_block_type(block_type):
-    block_type_table = Table(
+    block_type_table = rich.Table(
         title=block_type.name, show_header=False, show_footer=False, expand=True
     )
     block_type_table.add_column(style="italic cyan")
@@ -72,7 +71,9 @@ def display_block_type(block_type):
     return block_type_table
 
 
-async def _register_blocks_in_module(module: ModuleType) -> List[Type[Block]]:
+async def _register_blocks_in_module(module: ModuleType) -> List[Type["Block"]]:
+    from prefect.blocks.core import Block, InvalidBlockRegistration
+
     registered_blocks = []
     for _, cls in inspect.getmembers(module):
         if Block.is_block_class(cls):
@@ -85,8 +86,8 @@ async def _register_blocks_in_module(module: ModuleType) -> List[Type[Block]]:
     return registered_blocks
 
 
-def _build_registered_blocks_table(registered_blocks: List[Type[Block]]):
-    table = Table("Registered Blocks")
+def _build_registered_blocks_table(registered_blocks: List[Type["Block"]]):
+    table = rich.Table("Registered Blocks")
     for block in registered_blocks:
         table.add_row(block.get_block_type_name())
     return table
@@ -123,6 +124,13 @@ async def register(
         Register block types in a .py file:
         $ prefect block register -f my_blocks.py
     """
+    from importlib import import_module
+
+    from prefect.cli._utilities import exit_with_error
+    from prefect.settings import PREFECT_UI_URL
+    from prefect.utilities.asyncutils import run_sync_in_worker_thread
+    from prefect.utilities.importtools import load_script_as_module
+
     # Handles if both options are specified or if neither are specified
     if not (bool(file_path) ^ bool(module_name)):
         exit_with_error(
@@ -182,10 +190,12 @@ async def block_ls():
     """
     View all configured blocks.
     """
+    from prefect.client import get_client
+
     async with get_client() as client:
         blocks = await client.read_block_documents()
 
-    table = Table(
+    table = rich.Table(
         title="Blocks", caption="List Block Types using `prefect block type ls`"
     )
     table.add_column("ID", style="cyan", no_wrap=True)
@@ -214,6 +224,9 @@ async def block_delete(
     """
     Delete a configured block.
     """
+    from prefect.cli._utilities import exit_with_error, exit_with_success
+    from prefect.client import get_client
+
     async with get_client() as client:
         if slug is None and block_id is not None:
             try:
@@ -250,6 +263,10 @@ async def block_create(
     """
     Generate a link to the Prefect UI to create a block.
     """
+    from prefect.cli._utilities import exit_with_error
+    from prefect.client import get_client
+    from prefect.settings import PREFECT_UI_URL
+
     async with get_client() as client:
         try:
             block_type = await client.read_block_type_by_slug(block_type_slug)
@@ -284,6 +301,9 @@ async def block_inspect(
     """
     Displays details about a configured block.
     """
+    from prefect.cli._utilities import exit_with_error
+    from prefect.client import get_client
+
     async with get_client() as client:
         if slug is None and block_id is not None:
             try:
@@ -314,10 +334,12 @@ async def list_types():
     """
     List all block types.
     """
+    from prefect.client import get_client
+
     async with get_client() as client:
         block_types = await client.read_block_types()
 
-    table = Table(
+    table = rich.Table(
         title="Block Types",
         show_lines=True,
     )
@@ -349,6 +371,9 @@ async def blocktype_inspect(
     """
     Display details about a block type.
     """
+    from prefect.cli._utilities import exit_with_error
+    from prefect.client import get_client
+
     async with get_client() as client:
         try:
             block_type = await client.read_block_type_by_slug(slug)
@@ -365,6 +390,13 @@ async def blocktype_delete(
     """
     Delete an unprotected Block Type.
     """
+    from prefect.cli._utilities import exit_with_error, exit_with_success
+    from prefect.client import get_client
+    from prefect.exceptions import (
+        ObjectNotFound,
+        ProtectedBlockError,
+    )
+
     async with get_client() as client:
         try:
             block_type = await client.read_block_type_by_slug(slug)