let typing be compatible with python 3.8

src/integrations/prefect-dask/prefect_dask/task_runners.py

@@ -73,7 +73,7 @@ def count_to(highest_number):
 
 import inspect
 from contextlib import AsyncExitStack
-from typing import Awaitable, Callable, Dict, Optional, Union
+from typing import Callable, Dict, Optional, Union
 from uuid import UUID
 
 import distributed
@@ -82,7 +82,7 @@ def count_to(highest_number):
 from prefect.context import FlowRunContext
 from prefect.futures import PrefectFuture
 from prefect.states import exception_to_crashed_state
-from prefect.task_runners import BaseTaskRunner, R, TaskConcurrencyType
+from prefect.task_runners import BaseTaskRunner, CallableGeneric, R, TaskConcurrencyType
 from prefect.utilities.collections import visit_collection
 from prefect.utilities.importtools import from_qualified_name, to_qualified_name
 
@@ -254,7 +254,7 @@ def __eq__(self, other: object) -> bool:
     async def submit(
         self,
         key: UUID,
-        call: Callable[..., Awaitable[State[R]]],
+        call: CallableGeneric[R],
     ) -> None:
         if not self._started:
             raise RuntimeError(

src/integrations/prefect-ray/prefect_ray/task_runners.py

@@ -72,7 +72,7 @@ def count_to(highest_number):
 """
 
 from contextlib import AsyncExitStack
-from typing import Awaitable, Callable, Dict, Optional
+from typing import Dict, Optional
 from uuid import UUID
 
 import anyio
@@ -81,7 +81,7 @@ def count_to(highest_number):
 
 from prefect.futures import PrefectFuture
 from prefect.states import State, exception_to_crashed_state
-from prefect.task_runners import BaseTaskRunner, R, TaskConcurrencyType
+from prefect.task_runners import BaseTaskRunner, CallableGeneric, R, TaskConcurrencyType
 from prefect.utilities.asyncutils import sync_compatible
 from prefect.utilities.collections import visit_collection
 from prefect_ray.context import RemoteOptionsContext
@@ -153,7 +153,7 @@ def concurrency_type(self) -> TaskConcurrencyType:
     async def submit(
         self,
         key: UUID,
-        call: Callable[..., Awaitable[State[R]]],
+        call: CallableGeneric[R],
     ) -> None:
         if not self._started:
             raise RuntimeError(

src/prefect/task_runners.py

@@ -51,6 +51,7 @@
 
 For usage details, see the [Task Runners](/concepts/task-runners/) documentation.
 """
+
 import abc
 from contextlib import AsyncExitStack, asynccontextmanager
 from typing import (
@@ -60,6 +61,7 @@
     Awaitable,
     Callable,
     Dict,
+    Generic,
     Optional,
     Set,
     TypeVar,
@@ -80,6 +82,13 @@
 
 T = TypeVar("T", bound="BaseTaskRunner")
 R = TypeVar("R")
+StateR = TypeVar("StateR", bound=State[R])
+
+
+class CallableGeneric(Generic[R]):
+    # create this class to be compatible with python 3.8
+    def __call__(self, *args, **kwargs) -> Awaitable[StateR]:
+        pass
 
 
 class TaskConcurrencyType(AutoEnum):