Used anyio.abc full namespace instead of from imports

src/prefect/agent.py

@@ -6,9 +6,9 @@
 from uuid import UUID
 
 import anyio
+import anyio.abc
 import anyio.to_process
 import pendulum
-from anyio.abc import TaskGroup, TaskStatus
 
 from prefect.blocks.core import Block
 from prefect.client import OrionClient, get_client
@@ -41,7 +41,7 @@ def __init__(
         self.submitting_flow_run_ids = set()
         self.started = False
         self.logger = get_logger("agent")
-        self.task_group: Optional[TaskGroup] = None
+        self.task_group: Optional[anyio.abc.TaskGroup] = None
         self.client: Optional[OrionClient] = None
 
         self._work_queue_cache_expiration: pendulum.DateTime = None
@@ -214,7 +214,7 @@ async def _submit_run_and_capture_errors(
         self,
         flow_run: FlowRun,
         infrastructure: Infrastructure,
-        task_status: TaskStatus = None,
+        task_status: anyio.abc.TaskStatus = None,
     ) -> Union[InfrastructureResult, Exception]:
 
         # Note: There is not a clear way to determine if task_status.started() has been

src/prefect/context.py

@@ -23,8 +23,8 @@
     Union,
 )
 
+import anyio.abc
 import pendulum
-from anyio.abc import BlockingPortal, CancelScope, TaskGroup
 from pydantic import BaseModel, Field, PrivateAttr
 
 import prefect.logging
@@ -222,11 +222,11 @@ class FlowRunContext(RunContext):
 
     # The synchronous portal is only created for async flows for creating engine calls
     # from synchronous task and subflow calls
-    sync_portal: Optional[BlockingPortal] = None
-    timeout_scope: Optional[CancelScope] = None
+    sync_portal: Optional[anyio.abc.BlockingPortal] = None
+    timeout_scope: Optional[anyio.abc.CancelScope] = None
 
     # Task group that can be used for background tasks during the flow run
-    background_tasks: TaskGroup
+    background_tasks: anyio.abc.TaskGroup
 
     __var__ = ContextVar("flow_run")
 

src/prefect/infrastructure/base.py

@@ -1,8 +1,8 @@
 import abc
 from typing import TYPE_CHECKING, Dict, List, Optional
 
+import anyio.abc
 import pydantic
-from anyio.abc import TaskStatus
 from typing_extensions import Self
 
 import prefect
@@ -50,7 +50,7 @@ class Infrastructure(Block, abc.ABC):
     @abc.abstractmethod
     async def run(
         self,
-        task_status: TaskStatus = None,
+        task_status: anyio.abc.TaskStatus = None,
     ) -> InfrastructureResult:
         """
         Run the infrastructure.

src/prefect/infrastructure/docker.py

@@ -6,8 +6,8 @@
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, Dict, Generator, List, Optional, Tuple
 
+import anyio.abc
 import packaging.version
-from anyio.abc import TaskStatus
 from pydantic import Field, validator
 from slugify import slugify
 from typing_extensions import Literal
@@ -232,7 +232,7 @@ def check_volume_format(cls, volumes):
     @sync_compatible
     async def run(
         self,
-        task_status: Optional[TaskStatus] = None,
+        task_status: Optional[anyio.abc.TaskStatus] = None,
     ) -> Optional[bool]:
         if not self.command:
             raise ValueError("Docker container cannot be run with empty command.")

src/prefect/infrastructure/kubernetes.py

@@ -3,8 +3,8 @@
 from contextlib import contextmanager
 from typing import TYPE_CHECKING, Any, Dict, Generator, List, Optional, Union
 
+import anyio.abc
 import yaml
-from anyio.abc import TaskStatus
 from pydantic import Field, validator
 from slugify import slugify
 from typing_extensions import Literal
@@ -209,7 +209,7 @@ def customize_from_file(cls, filename: str) -> JsonPatch:
     @sync_compatible
     async def run(
         self,
-        task_status: Optional[TaskStatus] = None,
+        task_status: Optional[anyio.abc.TaskStatus] = None,
     ) -> Optional[bool]:
         if not self.command:
             raise ValueError("Kubernetes job cannot be run with empty command.")

src/prefect/infrastructure/process.py

@@ -4,8 +4,8 @@
 import tempfile
 from typing import Optional
 
+import anyio.abc
 import sniffio
-from anyio.abc import TaskStatus
 from pydantic import Field
 from typing_extensions import Literal
 
@@ -56,7 +56,7 @@ class Process(Infrastructure):
     @sync_compatible
     async def run(
         self,
-        task_status: TaskStatus = None,
+        task_status: anyio.abc.TaskStatus = None,
     ) -> Optional[bool]:
         if not self.command:
             raise ValueError("Process cannot be run with empty command.")

src/prefect/task_runners.py

@@ -68,7 +68,7 @@
 from prefect.utilities.collections import AutoEnum
 
 if TYPE_CHECKING:
-    from anyio.abc import TaskGroup
+    import anyio.abc
 
 from prefect.logging import get_logger
 from prefect.orion.schemas.states import State
@@ -229,7 +229,7 @@ def __init__(self):
         # TODO: Consider adding `max_workers` support using anyio capacity limiters
 
         # Runtime attributes
-        self._task_group: TaskGroup = None
+        self._task_group: anyio.abc.TaskGroup = None
         self._results: Dict[UUID, Any] = {}
         self._keys: Set[UUID] = set()
         super().__init__()