cleanup a few refs

src/prefect/client/orchestration/_deployments/client.py

@@ -7,7 +7,6 @@
 
 from prefect.client.orchestration.base import BaseAsyncClient, BaseClient
 from prefect.exceptions import ObjectNotFound
-from prefect.states import Scheduled
 
 if TYPE_CHECKING:
     import datetime
@@ -551,6 +550,7 @@ def create_flow_run_from_deployment(
         """
         from prefect.client.schemas.actions import DeploymentFlowRunCreate
         from prefect.client.schemas.objects import FlowRun
+        from prefect.states import Scheduled
 
         parameters = parameters or {}
         context = context or {}
@@ -1094,6 +1094,7 @@ async def create_flow_run_from_deployment(
         """
         from prefect.client.schemas.actions import DeploymentFlowRunCreate
         from prefect.client.schemas.objects import FlowRun
+        from prefect.states import Scheduled
 
         parameters = parameters or {}
         context = context or {}

src/prefect/client/orchestration/_logs/client.py

@@ -1,11 +1,8 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Iterable, Optional, Union
+from typing import TYPE_CHECKING, Any, Iterable, Union
 
 from prefect.client.orchestration.base import BaseAsyncClient, BaseClient
-from prefect.client.schemas.sorting import (
-    LogSort,
-)
 
 if TYPE_CHECKING:
     from prefect.client.schemas.actions import (
@@ -17,6 +14,7 @@
     from prefect.client.schemas.objects import (
         Log,
     )
+    from prefect.client.schemas.sorting import LogSort
 
 
 class LogClient(BaseClient):
@@ -34,19 +32,21 @@ def create_logs(self, logs: Iterable[Union["LogCreate", dict[str, Any]]]) -> Non
 
     def read_logs(
         self,
-        log_filter: Optional["LogFilter"] = None,
-        limit: Optional[int] = None,
-        offset: Optional[int] = None,
-        sort: "LogSort" = LogSort.TIMESTAMP_ASC,
+        log_filter: "LogFilter | None" = None,
+        limit: int | None = None,
+        offset: int | None = None,
+        sort: "LogSort | None" = None,
     ) -> list["Log"]:
         """
         Read flow and task run logs.
         """
+        from prefect.client.schemas.sorting import LogSort
+
         body: dict[str, Any] = {
             "logs": log_filter.model_dump(mode="json") if log_filter else None,
             "limit": limit,
             "offset": offset,
-            "sort": sort,
+            "sort": sort or LogSort.TIMESTAMP_ASC,
         }
         response = self.request("POST", "/logs/filter", json=body)
         from prefect.client.schemas.objects import Log
@@ -74,19 +74,21 @@ async def create_logs(
 
     async def read_logs(
         self,
-        log_filter: Optional["LogFilter"] = None,
-        limit: Optional[int] = None,
-        offset: Optional[int] = None,
-        sort: "LogSort" = LogSort.TIMESTAMP_ASC,
+        log_filter: "LogFilter | None" = None,
+        limit: int | None = None,
+        offset: int | None = None,
+        sort: "LogSort | None" = None,
     ) -> list[Log]:
         """
         Read flow and task run logs.
         """
+        from prefect.client.schemas.sorting import LogSort
+
         body: dict[str, Any] = {
             "logs": log_filter.model_dump(mode="json") if log_filter else None,
             "limit": limit,
             "offset": offset,
-            "sort": sort,
+            "sort": sort or LogSort.TIMESTAMP_ASC,
         }
 
         response = await self.request("POST", "/logs/filter", json=body)

src/prefect/client/schemas/__init__.py

@@ -1,32 +1,58 @@
-# Some objects are exported here for backwards compatibility.
-# In general, it is recommended to import schemas from their respective modules.
+import importlib
+import sys
+from typing import Any, TYPE_CHECKING
 
-from .actions import BlockTypeUpdate, StateCreate
-from .objects import (
-    DEFAULT_BLOCK_SCHEMA_VERSION,
-    BlockDocument,
-    BlockSchema,
-    BlockType,
-    FlowRun,
-    FlowRunPolicy,
-    State,
-    StateDetails,
-    StateType,
-    TaskRun,
-    TaskRunInput,
-    TaskRunPolicy,
-    TaskRunResult,
-    Workspace,
-)
-from .responses import (
-    OrchestrationResult,
-    SetStateStatus,
-    StateAbortDetails,
-    StateAcceptDetails,
-    StateRejectDetails,
-)
+if TYPE_CHECKING:
+    from .actions import BlockTypeUpdate, StateCreate
+    from .objects import (
+        DEFAULT_BLOCK_SCHEMA_VERSION,
+        BlockDocument,
+        BlockSchema,
+        BlockType,
+        FlowRun,
+        FlowRunPolicy,
+        State,
+        StateDetails,
+        StateType,
+        TaskRun,
+        TaskRunInput,
+        TaskRunPolicy,
+        TaskRunResult,
+        Workspace,
+    )
+    from .responses import (
+        OrchestrationResult,
+        SetStateStatus,
+        StateAbortDetails,
+        StateAcceptDetails,
+        StateRejectDetails,
+    )
 
-__all__ = (
+_public_api = {
+    "BlockDocument": (__package__, ".objects"),
+    "BlockSchema": (__package__, ".objects"),
+    "BlockType": (__package__, ".objects"),
+    "BlockTypeUpdate": (__package__, ".actions"),
+    "DEFAULT_BLOCK_SCHEMA_VERSION": (__package__, ".objects"),
+    "FlowRun": (__package__, ".objects"),
+    "FlowRunPolicy": (__package__, ".objects"),
+    "OrchestrationResult": (__package__, ".responses"),
+    "SetStateStatus": (__package__, ".responses"),
+    "State": (__package__, ".objects"),
+    "StateAbortDetails": (__package__, ".responses"),
+    "StateAcceptDetails": (__package__, ".responses"),
+    "StateCreate": (__package__, ".actions"),
+    "StateDetails": (__package__, ".objects"),
+    "StateRejectDetails": (__package__, ".responses"),
+    "StateType": (__package__, ".objects"),
+    "TaskRun": (__package__, ".objects"),
+    "TaskRunInput": (__package__, ".objects"),
+    "TaskRunPolicy": (__package__, ".objects"),
+    "TaskRunResult": (__package__, ".objects"),
+    "Workspace": (__package__, ".objects"),
+}
+
+__all__ = [
     "BlockDocument",
     "BlockSchema",
     "BlockType",
@@ -48,4 +74,18 @@
     "TaskRunPolicy",
     "TaskRunResult",
     "Workspace",
-)
+]
+
+
+def __getattr__(attr_name: str) -> Any:
+    try:
+        if (dynamic_attr := _public_api.get(attr_name)) is None:
+            raise AttributeError(f"module {__name__} has no attribute {attr_name}")
+
+        package, mname = dynamic_attr
+        module = importlib.import_module(mname, package=package)
+        return getattr(module, attr_name)
+    except ModuleNotFoundError as ex:
+        mname, _, attr = (ex.name or "").rpartition(".")
+        ctx = {"name": mname, "obj": attr} if sys.version_info >= (3, 10) else {}
+        raise AttributeError(f"module {mname} has no attribute {attr}", **ctx) from ex