Compatibility with 3.9 and small changes (#6)

Signed-off-by: Deepanshu Agarwal <deepanshu.agarwal1984@gmail.com>

Author: DeepanshuA

.github/workflows/pr-validation.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.11"]
+        python-version: ["3.9","3.10", "3.11"]
 
     steps:
     - uses: actions/checkout@v3

durabletask/client.py

@@ -6,7 +6,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
-from typing import Any, TypeVar
+from typing import Any, TypeVar, Union
 
 import grpc
 from google.protobuf import wrappers_pb2
@@ -42,10 +42,10 @@ class OrchestrationState:
     runtime_status: OrchestrationStatus
     created_at: datetime
     last_updated_at: datetime
-    serialized_input: str | None
-    serialized_output: str | None
-    serialized_custom_status: str | None
-    failure_details: task.FailureDetails | None
+    serialized_input: Union[str, None]
+    serialized_output: Union[str, None]
+    serialized_custom_status: Union[str, None]
+    failure_details: Union[task.FailureDetails, None]
 
     def raise_if_failed(self):
         if self.failure_details is not None:
@@ -64,7 +64,7 @@ def failure_details(self):
         return self._failure_details
 
 
-def new_orchestration_state(instance_id: str, res: pb.GetInstanceResponse) -> OrchestrationState | None:
+def new_orchestration_state(instance_id: str, res: pb.GetInstanceResponse) -> Union[OrchestrationState, None]:
     if not res.exists:
         return None
 
@@ -92,38 +92,39 @@ def new_orchestration_state(instance_id: str, res: pb.GetInstanceResponse) -> Or
 class TaskHubGrpcClient:
 
     def __init__(self, *,
-                 host_address: str | None = None,
+                 host_address: Union[str, None] = None,
                  log_handler=None,
-                 log_formatter: logging.Formatter | None = None):
+                 log_formatter: Union[logging.Formatter, None] = None):
         channel = shared.get_grpc_channel(host_address)
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
         self._logger = shared.get_logger(log_handler, log_formatter)
 
-    def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInput, TOutput], *,
-                                   input: TInput | None = None,
-                                   instance_id: str | None = None,
-                                   start_at: datetime | None = None) -> str:
+    def schedule_new_orchestration(self, orchestrator: Union[task.Orchestrator[TInput, TOutput], str], *,
+                                   input: Union[TInput, None] = None,
+                                   instance_id: Union[str, None] = None,
+                                   start_at: Union[datetime, None] = None) -> str:
 
-        name = task.get_name(orchestrator)
+        name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
 
         req = pb.CreateInstanceRequest(
             name=name,
             instanceId=instance_id if instance_id else uuid.uuid4().hex,
             input=wrappers_pb2.StringValue(value=shared.to_json(input)) if input else None,
-            scheduledStartTimestamp=helpers.new_timestamp(start_at) if start_at else None)
+            scheduledStartTimestamp=helpers.new_timestamp(start_at) if start_at else None,
+            version=wrappers_pb2.StringValue(value=""))
 
         self._logger.info(f"Starting new '{name}' instance with ID = '{req.instanceId}'.")
         res: pb.CreateInstanceResponse = self._stub.StartInstance(req)
         return res.instanceId
 
-    def get_orchestration_state(self, instance_id: str, *, fetch_payloads: bool = True) -> OrchestrationState | None:
+    def get_orchestration_state(self, instance_id: str, *, fetch_payloads: bool = True) -> Union[OrchestrationState, None]:
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         res: pb.GetInstanceResponse = self._stub.GetInstance(req)
         return new_orchestration_state(req.instanceId, res)
 
     def wait_for_orchestration_start(self, instance_id: str, *,
                                      fetch_payloads: bool = False,
-                                     timeout: int = 60) -> OrchestrationState | None:
+                                     timeout: int = 60) -> Union[OrchestrationState, None]:
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         try:
             self._logger.info(f"Waiting up to {timeout}s for instance '{instance_id}' to start.")
@@ -138,7 +139,7 @@ def wait_for_orchestration_start(self, instance_id: str, *,
 
     def wait_for_orchestration_completion(self, instance_id: str, *,
                                           fetch_payloads: bool = True,
-                                          timeout: int = 60) -> OrchestrationState | None:
+                                          timeout: int = 60) -> Union[OrchestrationState, None]:
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         try:
             self._logger.info(f"Waiting {timeout}s for instance '{instance_id}' to complete.")
@@ -152,7 +153,7 @@ def wait_for_orchestration_completion(self, instance_id: str, *,
                 raise
 
     def raise_orchestration_event(self, instance_id: str, event_name: str, *,
-                                  data: Any | None = None):
+                                  data: Union[Any, None] = None):
         req = pb.RaiseEventRequest(
             instanceId=instance_id,
             name=event_name,
@@ -162,7 +163,7 @@ def raise_orchestration_event(self, instance_id: str, event_name: str, *,
         self._stub.RaiseEvent(req)
 
     def terminate_orchestration(self, instance_id: str, *,
-                                output: Any | None = None):
+                                output: Union[Any, None] = None):
         req = pb.TerminateRequest(
             instanceId=instance_id,
             output=wrappers_pb2.StringValue(value=shared.to_json(output)) if output else None)

durabletask/internal/helpers.py

@@ -3,6 +3,7 @@
 
 import traceback
 from datetime import datetime
+from typing import Union
 
 from google.protobuf import timestamp_pb2, wrappers_pb2
 
@@ -11,14 +12,14 @@
 # TODO: The new_xxx_event methods are only used by test code and should be moved elsewhere
 
 
-def new_orchestrator_started_event(timestamp: datetime | None = None) -> pb.HistoryEvent:
+def new_orchestrator_started_event(timestamp: Union[datetime, None] = None) -> pb.HistoryEvent:
     ts = timestamp_pb2.Timestamp()
     if timestamp is not None:
         ts.FromDatetime(timestamp)
     return pb.HistoryEvent(eventId=-1, timestamp=ts, orchestratorStarted=pb.OrchestratorStartedEvent())
 
 
-def new_execution_started_event(name: str, instance_id: str, encoded_input: str | None = None) -> pb.HistoryEvent:
+def new_execution_started_event(name: str, instance_id: str, encoded_input: Union[str, None] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,
         timestamp=timestamp_pb2.Timestamp(),
@@ -48,15 +49,15 @@ def new_timer_fired_event(timer_id: int, fire_at: datetime) -> pb.HistoryEvent:
     )
 
 
-def new_task_scheduled_event(event_id: int, name: str, encoded_input: str | None = None) -> pb.HistoryEvent:
+def new_task_scheduled_event(event_id: int, name: str, encoded_input: Union[str, None] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=event_id,
         timestamp=timestamp_pb2.Timestamp(),
         taskScheduled=pb.TaskScheduledEvent(name=name, input=get_string_value(encoded_input))
     )
 
 
-def new_task_completed_event(event_id: int, encoded_output: str | None = None) -> pb.HistoryEvent:
+def new_task_completed_event(event_id: int, encoded_output: Union[str, None] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,
         timestamp=timestamp_pb2.Timestamp(),
@@ -76,7 +77,7 @@ def new_sub_orchestration_created_event(
         event_id: int,
         name: str,
         instance_id: str,
-        encoded_input: str | None = None) -> pb.HistoryEvent:
+        encoded_input: Union[str, None] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=event_id,
         timestamp=timestamp_pb2.Timestamp(),
@@ -87,7 +88,7 @@ def new_sub_orchestration_created_event(
     )
 
 
-def new_sub_orchestration_completed_event(event_id: int, encoded_output: str | None = None) -> pb.HistoryEvent:
+def new_sub_orchestration_completed_event(event_id: int, encoded_output: Union[str, None] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,
         timestamp=timestamp_pb2.Timestamp(),
@@ -115,7 +116,7 @@ def new_failure_details(ex: Exception) -> pb.TaskFailureDetails:
     )
 
 
-def new_event_raised_event(name: str, encoded_input: str | None = None) -> pb.HistoryEvent:
+def new_event_raised_event(name: str, encoded_input: Union[str, None] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,
         timestamp=timestamp_pb2.Timestamp(),
@@ -139,7 +140,7 @@ def new_resume_event() -> pb.HistoryEvent:
     )
 
 
-def new_terminated_event(*, encoded_output: str | None = None) -> pb.HistoryEvent:
+def new_terminated_event(*, encoded_output: Union[str, None] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,
         timestamp=timestamp_pb2.Timestamp(),
@@ -149,7 +150,7 @@ def new_terminated_event(*, encoded_output: str | None = None) -> pb.HistoryEven
     )
 
 
-def get_string_value(val: str | None) -> wrappers_pb2.StringValue | None:
+def get_string_value(val: Union[str, None]) -> Union[wrappers_pb2.StringValue, None]:
     if val is None:
         return None
     else:
@@ -159,8 +160,8 @@ def get_string_value(val: str | None) -> wrappers_pb2.StringValue | None:
 def new_complete_orchestration_action(
         id: int,
         status: pb.OrchestrationStatus,
-        result: str | None = None,
-        failure_details: pb.TaskFailureDetails | None = None) -> pb.OrchestratorAction:
+        result: Union[str, None] = None,
+        failure_details: Union[pb.TaskFailureDetails, None] = None) -> pb.OrchestratorAction:
 
     completeOrchestrationAction = pb.CompleteOrchestrationAction(
         orchestrationStatus=status,
@@ -178,7 +179,7 @@ def new_create_timer_action(id: int, fire_at: datetime) -> pb.OrchestratorAction
     return pb.OrchestratorAction(id=id, createTimer=pb.CreateTimerAction(fireAt=timestamp))
 
 
-def new_schedule_task_action(id: int, name: str, encoded_input: str | None) -> pb.OrchestratorAction:
+def new_schedule_task_action(id: int, name: str, encoded_input: Union[str, None]) -> pb.OrchestratorAction:
     return pb.OrchestratorAction(id=id, scheduleTask=pb.ScheduleTaskAction(
         name=name,
         input=get_string_value(encoded_input)
@@ -194,8 +195,8 @@ def new_timestamp(dt: datetime) -> timestamp_pb2.Timestamp:
 def new_create_sub_orchestration_action(
         id: int,
         name: str,
-        instance_id: str | None,
-        encoded_input: str | None) -> pb.OrchestratorAction:
+        instance_id: Union[str, None],
+        encoded_input: Union[str, None]) -> pb.OrchestratorAction:
     return pb.OrchestratorAction(id=id, createSubOrchestration=pb.CreateSubOrchestrationAction(
         name=name,
         instanceId=instance_id,

durabletask/internal/shared.py

@@ -5,7 +5,7 @@
 import json
 import logging
 from types import SimpleNamespace
-from typing import Any, Dict
+from typing import Any, Dict, Union
 
 import grpc
 
@@ -18,16 +18,16 @@ def get_default_host_address() -> str:
     return "localhost:4001"
 
 
-def get_grpc_channel(host_address: str | None) -> grpc.Channel:
+def get_grpc_channel(host_address: Union[str, None]) -> grpc.Channel:
     if host_address is None:
         host_address = get_default_host_address()
     channel = grpc.insecure_channel(host_address)
     return channel
 
 
 def get_logger(
-        log_handler: logging.Handler | None = None,
-        log_formatter: logging.Formatter | None = None) -> logging.Logger:
+        log_handler: Union[logging.Handler, None] = None,
+        log_formatter: Union[logging.Formatter, None] = None) -> logging.Logger:
     logger = logging.Logger("durabletask")
 
     # Add a default log handler if none is provided

durabletask/task.py

@@ -6,7 +6,7 @@
 
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
-from typing import Any, Callable, Generator, Generic, List, TypeVar
+from typing import Any, Callable, Generator, Generic, List, TypeVar, Union
 
 import durabletask.internal.helpers as pbh
 import durabletask.internal.orchestrator_service_pb2 as pb
@@ -70,7 +70,7 @@ def is_replaying(self) -> bool:
         pass
 
     @abstractmethod
-    def create_timer(self, fire_at: datetime | timedelta) -> Task:
+    def create_timer(self, fire_at: Union[datetime, timedelta]) -> Task:
         """Create a Timer Task to fire after at the specified deadline.
 
         Parameters
@@ -86,15 +86,15 @@ def create_timer(self, fire_at: datetime | timedelta) -> Task:
         pass
 
     @abstractmethod
-    def call_activity(self, activity: Activity[TInput, TOutput], *,
-                      input: TInput | None = None) -> Task[TOutput]:
+    def call_activity(self, activity: Union[Activity[TInput, TOutput], str], *,
+                      input: Union[TInput, None] = None) -> Task[TOutput]:
         """Schedule an activity for execution.
 
         Parameters
         ----------
-        name: Activity[TInput, TOutput]
+        activity: Union[Activity[TInput, TOutput], str]
             A reference to the activity function to call.
-        input: TInput | None
+        input: Union[TInput, None]
             The JSON-serializable input (or None) to pass to the activity.
         return_type: task.Task[TOutput]
             The JSON-serializable output type to expect from the activity result.
@@ -108,17 +108,17 @@ def call_activity(self, activity: Activity[TInput, TOutput], *,
 
     @abstractmethod
     def call_sub_orchestrator(self, orchestrator: Orchestrator[TInput, TOutput], *,
-                              input: TInput | None = None,
-                              instance_id: str | None = None) -> Task[TOutput]:
+                              input: Union[TInput, None] = None,
+                              instance_id: Union[str, None] = None) -> Task[TOutput]:
         """Schedule sub-orchestrator function for execution.
 
         Parameters
         ----------
         orchestrator: Orchestrator[TInput, TOutput]
             A reference to the orchestrator function to call.
-        input: TInput | None
+        input: Union[TInput, None]
             The optional JSON-serializable input to pass to the orchestrator function.
-        instance_id: str | None
+        instance_id: Union[str, None]
             A unique ID to use for the sub-orchestration instance. If not specified, a
             random UUID will be used.
 
@@ -149,7 +149,7 @@ def wait_for_external_event(self, name: str) -> Task:
 
 
 class FailureDetails:
-    def __init__(self, message: str, error_type: str, stack_trace: str | None):
+    def __init__(self, message: str, error_type: str, stack_trace: Union[str, None]):
         self._message = message
         self._error_type = error_type
         self._stack_trace = stack_trace
@@ -163,7 +163,7 @@ def error_type(self) -> str:
         return self._error_type
 
     @property
-    def stack_trace(self) -> str | None:
+    def stack_trace(self) -> Union[str, None]:
         return self._stack_trace
 
 
@@ -193,8 +193,8 @@ class OrchestrationStateError(Exception):
 class Task(ABC, Generic[T]):
     """Abstract base class for asynchronous tasks in a durable orchestration."""
     _result: T
-    _exception: TaskFailedError | None
-    _parent: CompositeTask[T] | None
+    _exception: Union[TaskFailedError, None]
+    _parent: Union[CompositeTask[T], None]
 
     def __init__(self) -> None:
         super().__init__()
@@ -357,7 +357,7 @@ def task_id(self) -> int:
 
 
 # Orchestrators are generators that yield tasks and receive/return any type
-Orchestrator = Callable[[OrchestrationContext, TInput], Generator[Task, Any, Any] | TOutput]
+Orchestrator = Callable[[OrchestrationContext, TInput], Union[Generator[Task, Any, Any], TOutput]]
 
 # Activities are simple functions that can be scheduled by orchestrators
 Activity = Callable[[ActivityContext, TInput], TOutput]