make py39 compatible,

Signed-off-by: Filinto Duran <[email protected]>

Author: filintod

dev-requirements.txt

@@ -1,6 +1,5 @@
 pytest-asyncio>=0.23
 pytest
 pytest-cov
-autopep8
 grpcio>=1.74.0
-protobuf>=6.31.1
+protobuf>=6.31.1  # TODO: move to pyproject.toml when tox v4 is used

durabletask/aio/awaitables.py

@@ -17,6 +17,7 @@
     Any,
     Awaitable,
     Callable,
+    Dict,
     Generator,
     Iterable,
     List,
@@ -80,8 +81,8 @@ def __init__(
         *,
         input: Any = None,
         retry_policy: Any = None,
-        app_id: str | None = None,
-        metadata: dict[str, str] | None = None,
+        app_id: Optional[str] = None,
+        metadata: Optional[Dict[str, str]] = None,
     ):
         """
         Initialize an activity awaitable.
@@ -115,7 +116,7 @@ def _to_task(self) -> task.Task[Any]:
             if (supports_metadata and self._metadata is not None) or (
                 supports_app_id and self._app_id is not None
             ):
-                kwargs: dict[str, Any] = {'input': self._input}
+                kwargs: Dict[str, Any] = {'input': self._input}
                 if supports_metadata and self._metadata is not None:
                     kwargs['metadata'] = self._metadata
                 if supports_app_id and self._app_id is not None:
@@ -129,7 +130,7 @@ def _to_task(self) -> task.Task[Any]:
             if (supports_metadata and self._metadata is not None) or (
                 supports_app_id and self._app_id is not None
             ):
-                kwargs2: dict[str, Any] = {'input': self._input, 'retry_policy': self._retry_policy}
+                kwargs2: Dict[str, Any] = {'input': self._input, 'retry_policy': self._retry_policy}
                 if supports_metadata and self._metadata is not None:
                     kwargs2['metadata'] = self._metadata
                 if supports_app_id and self._app_id is not None:
@@ -173,8 +174,8 @@ def __init__(
         input: Any = None,
         instance_id: Optional[str] = None,
         retry_policy: Any = None,
-        app_id: str | None = None,
-        metadata: dict[str, str] | None = None,
+        app_id: Optional[str] = None,
+        metadata: Optional[Dict[str, str]] = None,
     ):
         """
         Initialize a sub-orchestrator awaitable.
@@ -211,7 +212,7 @@ def _to_task(self) -> task.Task[Any]:
             if (supports_metadata and self._metadata is not None) or (
                 supports_app_id and self._app_id is not None
             ):
-                kwargs: dict[str, Any] = {'input': self._input, 'instance_id': self._instance_id}
+                kwargs: Dict[str, Any] = {'input': self._input, 'instance_id': self._instance_id}
                 if supports_metadata and self._metadata is not None:
                     kwargs['metadata'] = self._metadata
                 if supports_app_id and self._app_id is not None:
@@ -236,7 +237,7 @@ def _to_task(self) -> task.Task[Any]:
             if (supports_metadata and self._metadata is not None) or (
                 supports_app_id and self._app_id is not None
             ):
-                kwargs2: dict[str, Any] = {
+                kwargs2: Dict[str, Any] = {
                     'input': self._input,
                     'instance_id': self._instance_id,
                     'retry_policy': self._retry_policy,

durabletask/aio/compatibility.py

@@ -11,7 +11,7 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta
-from typing import Any, Optional, Protocol, Union, runtime_checkable
+from typing import Any, Dict, Optional, Protocol, Union, runtime_checkable
 
 from durabletask import task
 
@@ -92,7 +92,7 @@ def call_activity(
         *,
         input: Optional[Any] = None,
         retry_policy: Optional[task.RetryPolicy] = None,
-        metadata: Optional[dict[str, str]] = None,
+        metadata: Optional[Dict[str, str]] = None,
     ) -> Any:
         """Schedule an activity for execution."""
         ...
@@ -104,7 +104,7 @@ def call_sub_orchestrator(
         input: Optional[Any] = None,
         instance_id: Optional[str] = None,
         retry_policy: Optional[task.RetryPolicy] = None,
-        metadata: Optional[dict[str, str]] = None,
+        metadata: Optional[Dict[str, str]] = None,
     ) -> Any:
         """Schedule sub-orchestrator function for execution."""
         ...

durabletask/aio/context.py

@@ -98,7 +98,7 @@ def __init__(self, base_ctx: dt_task.OrchestrationContext):
         self._workflow_name: Optional[str] = None
         self._current_step: Optional[str] = None
         # Set by sandbox when active
-        self._sandbox_originals: Optional[dict[str, Any]] = None
+        self._sandbox_originals: Optional[Dict[str, Any]] = None
         self._sandbox_mode: Optional[str] = None
 
         # Performance optimization: Check if detection should be globally disabled
@@ -236,7 +236,7 @@ def activity(
         input: Any = None,
         retry_policy: Any = None,
         app_id: Optional[str] = None,
-        metadata: dict[str, str] | None = None,
+        metadata: Optional[Dict[str, str]] = None,
     ) -> ActivityAwaitable[Any]:
         """
         Create an awaitable for calling an activity function.
@@ -267,7 +267,7 @@ def call_activity(
         input: Any = None,
         retry_policy: Any = None,
         app_id: Optional[str] = None,
-        metadata: dict[str, str] | None = None,
+        metadata: Optional[Dict[str, str]] = None,
     ) -> ActivityAwaitable[Any]:
         """Alias for activity() method for API compatibility."""
         return self.activity(
@@ -287,7 +287,7 @@ def sub_orchestrator(
         instance_id: Optional[str] = None,
         retry_policy: Any = None,
         app_id: Optional[str] = None,
-        metadata: dict[str, str] | None = None,
+        metadata: Optional[Dict[str, str]] = None,
     ) -> SubOrchestratorAwaitable[Any]:
         """
         Create an awaitable for calling a sub-orchestrator.
@@ -324,7 +324,7 @@ def call_sub_orchestrator(
         instance_id: Optional[str] = None,
         retry_policy: Any = None,
         app_id: Optional[str] = None,
-        metadata: dict[str, str] | None = None,
+        metadata: Optional[Dict[str, str]] = None,
     ) -> SubOrchestratorAwaitable[Any]:
         """Call a sub-orchestrator workflow (durabletask naming convention)."""
         return self.sub_orchestrator(
@@ -472,7 +472,7 @@ def continue_as_new(self, input_data: Any = None, *, save_events: bool = False)
                 self._base_ctx.continue_as_new(input_data, save_events=save_events)
 
     # Metadata and header methods
-    def set_metadata(self, metadata: dict[str, str]) -> None:
+    def set_metadata(self, metadata: Dict[str, str]) -> None:
         """
         Set metadata for the workflow instance.
 
@@ -483,7 +483,7 @@ def set_metadata(self, metadata: dict[str, str]) -> None:
             self._base_ctx.set_metadata(metadata)
         self._log_operation('set_metadata', {'metadata': metadata})
 
-    def get_metadata(self) -> Optional[dict[str, str]]:
+    def get_metadata(self) -> Optional[Dict[str, str]]:
         """
         Get metadata for the workflow instance.
 
@@ -493,10 +493,10 @@ def get_metadata(self) -> Optional[dict[str, str]]:
         if hasattr(self._base_ctx, 'get_metadata'):
             val: Any = self._base_ctx.get_metadata()
             if isinstance(val, dict):
-                return cast(dict[str, str], val)
+                return cast(Dict[str, str], val)
         return None
 
-    def set_headers(self, headers: dict[str, str]) -> None:
+    def set_headers(self, headers: Dict[str, str]) -> None:
         """
         Set headers for the workflow instance (alias for set_metadata).
 
@@ -505,7 +505,7 @@ def set_headers(self, headers: dict[str, str]) -> None:
         """
         self.set_metadata(headers)
 
-    def get_headers(self) -> Optional[dict[str, str]]:
+    def get_headers(self) -> Optional[Dict[str, str]]:
         """
         Get headers for the workflow instance (alias for get_metadata).
 

durabletask/aio/sandbox.py

@@ -20,7 +20,7 @@
 from datetime import timedelta
 from enum import Enum
 from types import FrameType
-from typing import Any, Callable, Optional, Union, cast
+from typing import Any, Callable, Dict, Optional, Set, Type, Union, cast
 
 from durabletask.deterministic import deterministic_random, deterministic_uuid4
 
@@ -38,17 +38,17 @@ class SandboxMode(str, Enum):
     STRICT = 'strict'
 
 
-def _as_mode_str(mode: str | SandboxMode) -> str:
+def _as_mode_str(mode: Union[str, SandboxMode]) -> str:
     return mode.value if isinstance(mode, SandboxMode) else mode
 
 
 class _NonDeterminismDetector:
     """Detects and warns about non-deterministic function calls in workflows."""
 
-    def __init__(self, async_ctx: Any, mode: str | SandboxMode):
+    def __init__(self, async_ctx: Any, mode: Union[str, SandboxMode]):
         self.async_ctx = async_ctx
         self.mode = _as_mode_str(mode)
-        self.detected_calls: set[str] = set()
+        self.detected_calls: Set[str] = set()
         self.original_trace_func: Optional[Callable[[FrameType, str, Any], Any]] = None
         self._restore_trace_func: Optional[Callable[[FrameType, str, Any], Any]] = None
         self._active_trace_func: Optional[Callable[[FrameType, str, Any], Any]] = None
@@ -74,7 +74,7 @@ def __enter__(self) -> '_NonDeterminismDetector':
 
     def __exit__(
         self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
         exc_val: Optional[BaseException],
         exc_tb: Optional[Any],
     ) -> None:
@@ -272,11 +272,11 @@ def _get_deterministic_alternative(self, function_name: str) -> str:
 class _Sandbox(ContextDecorator):
     """Context manager for sandboxing workflow execution."""
 
-    def __init__(self, async_ctx: Any, mode: str | SandboxMode):
+    def __init__(self, async_ctx: Any, mode: Union[str, SandboxMode]):
         self.async_ctx = async_ctx
         self.mode = _as_mode_str(mode)
-        self.originals: dict[str, Any] = {}
-        self.detector: _NonDeterminismDetector | None = None
+        self.originals: Dict[str, Any] = {}
+        self.detector: Optional[_NonDeterminismDetector] = None
 
     def __enter__(self) -> '_Sandbox':
         if self.mode == 'off':
@@ -305,7 +305,7 @@ def __enter__(self) -> '_Sandbox':
 
     def __exit__(
         self,
-        exc_type: Optional[type[BaseException]],
+        exc_type: Optional[Type[BaseException]],
         exc_val: Optional[BaseException],
         exc_tb: Optional[Any],
     ) -> None:
@@ -492,7 +492,7 @@ def __init__(self, factory: Callable[[], Any]) -> None:
                 self._factory = factory
                 self._done = False
                 self._res: Any = None
-                self._exc: BaseException | None = None
+                self._exc: Optional[BaseException] = None
 
             def __await__(self) -> Any:
                 if self._done:
@@ -717,7 +717,7 @@ def _restore_originals(self) -> None:
 
 
 @contextlib.contextmanager
-def sandbox_scope(async_ctx: Any, mode: str | SandboxMode) -> Any:
+def sandbox_scope(async_ctx: Any, mode: Union[str, SandboxMode]) -> Any:
     """
     Create a sandbox context for deterministic workflow execution.
 

durabletask/deterministic.py

@@ -17,7 +17,7 @@
 import uuid
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Protocol, Sequence, TypeVar, runtime_checkable
+from typing import Optional, Protocol, Sequence, TypeVar, runtime_checkable
 
 
 @dataclass
@@ -105,7 +105,7 @@ def new_guid(self) -> uuid.UUID:
         """Alias for uuid4 for API parity with other SDKs."""
         return self.uuid4()
 
-    def random_string(self, length: int, *, alphabet: str | None = None) -> str:
+    def random_string(self, length: int, *, alphabet: Optional[str] = None) -> str:
         """Return a deterministically generated random string of the given length."""
         if length < 0:
             raise ValueError('length must be non-negative')

durabletask/internal/shared.py

@@ -6,7 +6,7 @@
 import logging
 import os
 from types import SimpleNamespace
-from typing import Any, Iterable, Optional, Sequence, Union
+from typing import Any, Dict, Iterable, Optional, Sequence, Union
 
 import grpc
 
@@ -274,7 +274,7 @@ class InternalJSONDecoder(json.JSONDecoder):
     def __init__(self, *args, **kwargs):
         super().__init__(object_hook=self.dict_to_object, *args, **kwargs)
 
-    def dict_to_object(self, d: dict[str, Any]):
+    def dict_to_object(self, d: Dict[str, Any]):
         # If the object was serialized by the InternalJSONEncoder, deserialize it as a SimpleNamespace
         if d.pop(AUTO_SERIALIZED, False):
             return SimpleNamespace(**d)

durabletask/worker.py

@@ -10,7 +10,7 @@
 from datetime import datetime, timedelta
 from threading import Event, Thread
 from types import GeneratorType
-from typing import Any, Callable, Generator, Optional, Sequence, TypeVar, Union
+from typing import Any, Callable, Dict, Generator, Optional, Sequence, TypeVar, Union
 
 import grpc
 from google.protobuf import empty_pb2
@@ -74,8 +74,8 @@ def __init__(
 
 
 class _Registry:
-    orchestrators: dict[str, task.Orchestrator]
-    activities: dict[str, task.Activity]
+    orchestrators: Dict[str, task.Orchestrator]
+    activities: Dict[str, task.Activity]
 
     def __init__(self):
         self.orchestrators = {}
@@ -808,15 +808,15 @@ def __init__(self, instance_id: str):
         self._is_suspended = False
         self._is_complete = False
         self._result = None
-        self._pending_actions: dict[int, pb.OrchestratorAction] = {}
-        self._pending_tasks: dict[int, task.CompletableTask] = {}
+        self._pending_actions: Dict[int, pb.OrchestratorAction] = {}
+        self._pending_tasks: Dict[int, task.CompletableTask] = {}
         self._sequence_number = 0
         self._current_utc_datetime = datetime(1000, 1, 1)
         self._instance_id = instance_id
         self._app_id = None
         self._completion_status: Optional[pb.OrchestrationStatus] = None
-        self._received_events: dict[str, list[Any]] = {}
-        self._pending_events: dict[str, list[task.CompletableTask]] = {}
+        self._received_events: Dict[str, list[Any]] = {}
+        self._pending_events: Dict[str, list[task.CompletableTask]] = {}
         self._new_input: Optional[Any] = None
         self._save_events = False
         self._encoded_custom_status: Optional[str] = None
@@ -1757,7 +1757,7 @@ def _get_new_event_summary(new_events: Sequence[pb.HistoryEvent]) -> str:
     elif len(new_events) == 1:
         return f'[{new_events[0].WhichOneof("eventType")}]'
     else:
-        counts: dict[str, int] = {}
+        counts: Dict[str, int] = {}
         for event in new_events:
             event_type = event.WhichOneof('eventType')
             counts[event_type] = counts.get(event_type, 0) + 1
@@ -1771,7 +1771,7 @@ def _get_action_summary(new_actions: Sequence[pb.OrchestratorAction]) -> str:
     elif len(new_actions) == 1:
         return f'[{new_actions[0].WhichOneof("orchestratorActionType")}]'
     else:
-        counts: dict[str, int] = {}
+        counts: Dict[str, int] = {}
         for action in new_actions:
             action_type = action.WhichOneof('orchestratorActionType')
             counts[action_type] = counts.get(action_type, 0) + 1

requirements.txt

@@ -1 +1 @@
-
+# pyproject.toml has the dependencies for this project