Fix all typing errors in prefect.runner.runner (#16797)

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

@@ -1031,8 +1031,8 @@ async def get_scheduled_flow_runs_for_deployments(
         deployment_ids: list["UUID"],
         scheduled_before: "datetime.datetime | None" = None,
         limit: int | None = None,
-    ) -> list["FlowRunResponse"]:
-        from prefect.client.schemas.responses import FlowRunResponse
+    ) -> list["FlowRun"]:
+        from prefect.client.schemas.objects import FlowRun
 
         body: dict[str, Any] = dict(deployment_ids=[str(id) for id in deployment_ids])
         if scheduled_before:
@@ -1046,7 +1046,7 @@ async def get_scheduled_flow_runs_for_deployments(
             json=body,
         )
 
-        return FlowRunResponse.model_validate_list(response.json())
+        return FlowRun.model_validate_list(response.json())
 
     async def create_flow_run_from_deployment(
         self,

src/prefect/flows.py

@@ -127,9 +127,9 @@ def __call__(
     import logging
 
     from prefect.client.orchestration import PrefectClient
+    from prefect.client.schemas.objects import FlowRun
     from prefect.client.types.flexible_schedule_list import FlexibleScheduleList
     from prefect.deployments.runner import RunnerDeployment
-    from prefect.flows import FlowRun
     from prefect.runner.storage import RunnerStorage
 
 logger: "logging.Logger" = get_logger("flows")

src/prefect/logging/loggers.py

@@ -24,7 +24,6 @@
         LoggingAdapter = logging.LoggerAdapter
 
 if TYPE_CHECKING:
-    from prefect.client.schemas import FlowRun as ClientFlowRun
     from prefect.client.schemas.objects import FlowRun, TaskRun
     from prefect.context import RunContext
     from prefect.flows import Flow
@@ -164,7 +163,7 @@ def get_run_logger(
 
 
 def flow_run_logger(
-    flow_run: Union["FlowRun", "ClientFlowRun"],
+    flow_run: "FlowRun",
     flow: Optional["Flow[Any, Any]"] = None,
     **kwargs: str,
 ) -> PrefectLogAdapter: