Use pre-PEP585 types for compatibility with 3.8

Author: chriso

src/dispatch/coroutine.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from types import coroutine
-from typing import Any, Awaitable
+from typing import Any, Awaitable, List, Tuple
 
 from dispatch.experimental.durable import durable
 from dispatch.proto import Call
@@ -16,14 +16,14 @@ def call(call: Call) -> Any:
 
 @coroutine
 @durable
-def gather(*awaitables: Awaitable[Any]) -> list[Any]:  # type: ignore[misc]
+def gather(*awaitables: Awaitable[Any]) -> List[Any]:  # type: ignore[misc]
     """Alias for all."""
     return all(*awaitables)
 
 
 @coroutine
 @durable
-def all(*awaitables: Awaitable[Any]) -> list[Any]:  # type: ignore[misc]
+def all(*awaitables: Awaitable[Any]) -> List[Any]:  # type: ignore[misc]
     """Concurrently run a set of coroutines, blocking until all coroutines
     return or any coroutine raises an error. If any coroutine fails with an
     uncaught exception, the exception will be re-raised here."""
@@ -32,7 +32,7 @@ def all(*awaitables: Awaitable[Any]) -> list[Any]:  # type: ignore[misc]
 
 @coroutine
 @durable
-def any(*awaitables: Awaitable[Any]) -> list[Any]:  # type: ignore[misc]
+def any(*awaitables: Awaitable[Any]) -> List[Any]:  # type: ignore[misc]
     """Concurrently run a set of coroutines, blocking until any coroutine
     returns or all coroutines raises an error. If all coroutines fail with
     uncaught exceptions, the exception(s) will be re-raised here."""
@@ -41,7 +41,7 @@ def any(*awaitables: Awaitable[Any]) -> list[Any]:  # type: ignore[misc]
 
 @coroutine
 @durable
-def race(*awaitables: Awaitable[Any]) -> list[Any]:  # type: ignore[misc]
+def race(*awaitables: Awaitable[Any]) -> List[Any]:  # type: ignore[misc]
     """Concurrently run a set of coroutines, blocking until any coroutine
     returns or raises an error. If any coroutine fails with an uncaught
     exception, the exception will be re-raised here."""
@@ -50,17 +50,17 @@ def race(*awaitables: Awaitable[Any]) -> list[Any]:  # type: ignore[misc]
 
 @dataclass
 class AllDirective:
-    awaitables: tuple[Awaitable[Any], ...]
+    awaitables: Tuple[Awaitable[Any], ...]
 
 
 @dataclass
 class AnyDirective:
-    awaitables: tuple[Awaitable[Any], ...]
+    awaitables: Tuple[Awaitable[Any], ...]
 
 
 @dataclass
 class RaceDirective:
-    awaitables: tuple[Awaitable[Any], ...]
+    awaitables: Tuple[Awaitable[Any], ...]
 
 
 class AnyException(RuntimeError):
@@ -69,7 +69,7 @@ class AnyException(RuntimeError):
 
     __slots__ = ("exceptions",)
 
-    def __init__(self, exceptions: list[Exception]):
+    def __init__(self, exceptions: List[Exception]):
         self.exceptions = exceptions
 
     def __str__(self):

src/dispatch/experimental/durable/function.py

@@ -9,7 +9,7 @@
     MethodType,
     TracebackType,
 )
-from typing import Any, Callable, Coroutine, Generator, Optional, TypeVar, Union, cast
+from typing import Any, Callable, Coroutine, Generator, Optional, TypeVar, Union, cast, Dict, Tuple
 
 from . import frame as ext
 from .registry import RegisteredFunction, lookup_function, register_function
@@ -78,8 +78,8 @@ class Serializable:
     g: Union[GeneratorType, CoroutineType]
     registered_fn: RegisteredFunction
     wrapped_coroutine: Union["DurableCoroutine", None]
-    args: tuple[Any, ...]
-    kwargs: dict[str, Any]
+    args: Tuple[Any, ...]
+    kwargs: Dict[str, Any]
 
     def __init__(
         self,
@@ -274,7 +274,7 @@ def cr_await(self) -> Any:
         return self.coroutine.cr_await
 
     @property
-    def cr_origin(self) -> Optional[tuple[tuple[str, int, str], ...]]:
+    def cr_origin(self) -> Optional[Tuple[Tuple[str, int, str], ...]]:
         return self.coroutine.cr_origin
 
     def __repr__(self) -> str:

src/dispatch/experimental/durable/registry.py

@@ -1,6 +1,7 @@
 import hashlib
 from dataclasses import dataclass
 from types import FunctionType
+from typing import Dict
 
 
 @dataclass
@@ -46,7 +47,7 @@ def __setstate__(self, state):
         self.hash = code_hash
 
 
-_REGISTRY: dict[str, RegisteredFunction] = {}
+_REGISTRY: Dict[str, RegisteredFunction] = {}
 
 
 def register_function(fn: FunctionType) -> RegisteredFunction:

src/dispatch/function.py

@@ -15,6 +15,7 @@
     Optional,
     TypeVar,
     overload,
+    List
 )
 from urllib.parse import urlparse
 
@@ -329,7 +330,7 @@ def batch(self) -> Batch:
         a set of calls to dispatch."""
         return Batch(self)
 
-    def dispatch(self, calls: Iterable[Call]) -> list[DispatchID]:
+    def dispatch(self, calls: Iterable[Call]) -> List[DispatchID]:
         """Dispatch function calls.
 
         Args:
@@ -369,7 +370,7 @@ class Batch:
 
     def __init__(self, client: Client):
         self.client = client
-        self.calls: list[Call] = []
+        self.calls: List[Call] = []
 
     def add(self, func: Function[P, T], *args: P.args, **kwargs: P.kwargs) -> Batch:
         """Add a call to the specified function to the batch."""
@@ -380,7 +381,7 @@ def add_call(self, call: Call) -> Batch:
         self.calls.append(call)
         return self
 
-    def dispatch(self) -> list[DispatchID]:
+    def dispatch(self) -> List[DispatchID]:
         """Dispatch dispatches the calls asynchronously.
 
         The batch is reset when the calls are dispatched successfully.

src/dispatch/proto.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from traceback import format_exception
 from types import TracebackType
-from typing import Any, Optional
+from typing import Any, Optional, Dict, Tuple, List
 
 import google.protobuf.any_pb2
 import google.protobuf.message
@@ -79,7 +79,7 @@ def input(self) -> Any:
         self._assert_first_call()
         return self._input
 
-    def input_arguments(self) -> tuple[tuple[Any, ...], dict[str, Any]]:
+    def input_arguments(self) -> Tuple[Tuple[Any, ...], Dict[str, Any]]:
         """Returns positional and keyword arguments carried by the input."""
         self._assert_first_call()
         if not isinstance(self._input, Arguments):
@@ -92,7 +92,7 @@ def coroutine_state(self) -> Any:
         return self._coroutine_state
 
     @property
-    def call_results(self) -> list[CallResult]:
+    def call_results(self) -> List[CallResult]:
         self._assert_resume()
         return self._call_results
 
@@ -124,7 +124,7 @@ def from_poll_results(
         cls,
         function: str,
         coroutine_state: Any,
-        call_results: list[CallResult],
+        call_results: List[CallResult],
         error: Optional[Error] = None,
     ):
         return Input(
@@ -143,8 +143,8 @@ def from_poll_results(
 class Arguments:
     """A container for positional and keyword arguments."""
 
-    args: tuple[Any, ...]
-    kwargs: dict[str, Any]
+    args: Tuple[Any, ...]
+    kwargs: Dict[str, Any]
 
 
 @dataclass
@@ -201,7 +201,7 @@ def exit(
     def poll(
         cls,
         state: Any,
-        calls: Optional[list[Call]] = None,
+        calls: Optional[List[Call]] = None,
         min_results: int = 1,
         max_results: int = 10,
         max_wait_seconds: Optional[int] = None,

src/dispatch/scheduler.py

@@ -2,7 +2,7 @@
 import pickle
 import sys
 from dataclasses import dataclass, field
-from typing import Any, Awaitable, Callable, Optional, Protocol, Union
+from typing import Any, Awaitable, Callable, Optional, Protocol, Union, List, Dict, Tuple, Set
 
 from typing_extensions import TypeAlias
 
@@ -84,9 +84,9 @@ def value(self) -> Any:
 class AllFuture:
     """A future result of a dispatch.coroutine.all() operation."""
 
-    order: list[CoroutineID] = field(default_factory=list)
-    waiting: set[CoroutineID] = field(default_factory=set)
-    results: dict[CoroutineID, CoroutineResult] = field(default_factory=dict)
+    order: List[CoroutineID] = field(default_factory=list)
+    waiting: Set[CoroutineID] = field(default_factory=set)
+    results: Dict[CoroutineID, CoroutineResult] = field(default_factory=dict)
     first_error: Optional[Exception] = None
 
     def add_result(self, result: Union[CallResult, CoroutineResult]):
@@ -115,7 +115,7 @@ def error(self) -> Optional[Exception]:
         assert self.ready()
         return self.first_error
 
-    def value(self) -> list[Any]:
+    def value(self) -> List[Any]:
         assert self.ready()
         assert len(self.waiting) == 0
         assert self.first_error is None
@@ -126,10 +126,10 @@ def value(self) -> list[Any]:
 class AnyFuture:
     """A future result of a dispatch.coroutine.any() operation."""
 
-    order: list[CoroutineID] = field(default_factory=list)
-    waiting: set[CoroutineID] = field(default_factory=set)
+    order: List[CoroutineID] = field(default_factory=list)
+    waiting: Set[CoroutineID] = field(default_factory=set)
     first_result: Optional[CoroutineResult] = None
-    errors: dict[CoroutineID, Exception] = field(default_factory=dict)
+    errors: Dict[CoroutineID, Exception] = field(default_factory=dict)
     generic_error: Optional[Exception] = None
 
     def add_result(self, result: Union[CallResult, CoroutineResult]):
@@ -183,7 +183,7 @@ def value(self) -> Any:
 class RaceFuture:
     """A future result of a dispatch.coroutine.race() operation."""
 
-    waiting: set[CoroutineID] = field(default_factory=set)
+    waiting: Set[CoroutineID] = field(default_factory=set)
     first_result: Optional[CoroutineResult] = None
     first_error: Optional[Exception] = None
 
@@ -248,12 +248,12 @@ class State:
     """State of the scheduler and the coroutines it's managing."""
 
     version: str
-    suspended: dict[CoroutineID, Coroutine]
-    ready: list[Coroutine]
+    suspended: Dict[CoroutineID, Coroutine]
+    ready: List[Coroutine]
     next_coroutine_id: int
     next_call_id: int
 
-    prev_callers: list[Coroutine]
+    prev_callers: List[Coroutine]
 
     outstanding_calls: int
 
@@ -416,7 +416,7 @@ def _run(self, input: Input) -> Output:
             len(state.ready) + len(state.suspended),
         )
 
-        pending_calls: list[Call] = []
+        pending_calls: List[Call] = []
         while state.ready:
             coroutine = state.ready.pop(0)
             logger.debug("running %s", coroutine)
@@ -542,8 +542,8 @@ def _run(self, input: Input) -> Output:
 
 
 def spawn_children(
-    state: State, coroutine: Coroutine, awaitables: tuple[Awaitable[Any], ...]
-) -> list[Coroutine]:
+    state: State, coroutine: Coroutine, awaitables: Tuple[Awaitable[Any], ...]
+) -> List[Coroutine]:
     children = []
     for awaitable in awaitables:
         g = awaitable.__await__()

src/dispatch/signature/__init__.py

@@ -1,6 +1,6 @@
 import logging
 from datetime import datetime, timedelta
-from typing import Sequence, cast
+from typing import Sequence, cast, Set
 
 import http_sfv
 from cryptography.hazmat.primitives.asymmetric.ed25519 import (
@@ -114,8 +114,8 @@ def verify_request(request: Request, key: Ed25519PublicKey, max_age: timedelta):
     verify_content_digest(request.headers["Content-Digest"], request.body)
 
 
-def extract_covered_components(result: VerifyResult) -> set[str]:
-    covered_components: set[str] = set()
+def extract_covered_components(result: VerifyResult) -> Set[str]:
+    covered_components: Set[str] = set()
     for key in result.covered_components.keys():
         item = http_sfv.Item()
         item.parse(key.encode())

src/dispatch/status.py

@@ -1,5 +1,5 @@
 import enum
-from typing import Any, Callable, Type
+from typing import Any, Callable, Type, Dict
 
 from dispatch.error import IncompatibleStateError
 from dispatch.sdk.v1 import status_pb2 as status_pb
@@ -79,8 +79,8 @@ def __str__(self):
 Status.NOT_FOUND.__doc__ = "An operation was performed on a non-existent resource"
 Status.NOT_FOUND._proto = status_pb.STATUS_NOT_FOUND
 
-_ERROR_TYPES: dict[Type[Exception], Callable[[Exception], Status]] = {}
-_OUTPUT_TYPES: dict[Type[Any], Callable[[Any], Status]] = {}
+_ERROR_TYPES: Dict[Type[Exception], Callable[[Exception], Status]] = {}
+_OUTPUT_TYPES: Dict[Type[Any], Callable[[Any], Status]] = {}
 
 
 def status_for_error(error: Exception) -> Status:

src/dispatch/test/service.py

@@ -5,7 +5,7 @@
 import time
 from collections import OrderedDict
 from dataclasses import dataclass
-from typing import Optional
+from typing import Optional, Dict, Tuple, List, Set
 
 import grpc
 import httpx
@@ -34,7 +34,7 @@
 logger = logging.getLogger(__name__)
 
 
-RoundTrip: TypeAlias = tuple[function_pb.RunRequest, function_pb.RunResponse]
+RoundTrip: TypeAlias = Tuple[function_pb.RunRequest, function_pb.RunResponse]
 """A request to a Dispatch endpoint, and the response that was received."""
 
 
@@ -54,7 +54,7 @@ def __init__(
         self,
         endpoint_client: EndpointClient,
         api_key: Optional[str] = None,
-        retry_on_status: Optional[set[Status]] = None,
+        retry_on_status: Optional[Set[Status]] = None,
         collect_roundtrips: bool = False,
     ):
         """Initialize the Dispatch service.
@@ -82,12 +82,12 @@ def __init__(
 
         self._next_dispatch_id = 1
 
-        self.queue: list[tuple[DispatchID, function_pb.RunRequest, CallType]] = []
+        self.queue: List[Tuple[DispatchID, function_pb.RunRequest, CallType]] = []
 
-        self.pollers: dict[DispatchID, Poller] = {}
-        self.parents: dict[DispatchID, Poller] = {}
+        self.pollers: Dict[DispatchID, Poller] = {}
+        self.parents: Dict[DispatchID, Poller] = {}
 
-        self.roundtrips: Optional[OrderedDict[DispatchID, list[RoundTrip]]] = None
+        self.roundtrips: Optional[OrderedDict[DispatchID, List[RoundTrip]]] = None
         if collect_roundtrips:
             self.roundtrips = OrderedDict()
 
@@ -354,5 +354,5 @@ class Poller:
     coroutine_state: bytes
     # TODO: support max_wait/min_results/max_results
 
-    waiting: dict[DispatchID, call_pb.Call]
-    results: dict[DispatchID, call_pb.CallResult]
+    waiting: Dict[DispatchID, call_pb.Call]
+    results: Dict[DispatchID, call_pb.CallResult]