Adjusted implementation per dapr version

linting

Author: passuied

durabletask/aio/client.py

@@ -24,8 +24,7 @@ def __init__(self, *,
                  log_handler: Optional[logging.Handler] = None,
                  log_formatter: Optional[logging.Formatter] = None,
                  secure_channel: bool = False,
-                 interceptors: Optional[Sequence[AioClientInterceptor]] = None,
-                 default_version: Optional[str] = None):
+                 interceptors: Optional[Sequence[AioClientInterceptor]] = None):
 
         if interceptors is not None:
             interceptors = list(interceptors)
@@ -44,7 +43,6 @@ def __init__(self, *,
         self._channel = channel
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
         self._logger = shared.get_logger("client", log_handler, log_formatter)
-        self.default_version = default_version
 
     async def aclose(self):
         await self._channel.close()
@@ -53,9 +51,7 @@ async def schedule_new_orchestration(self, orchestrator: Union[task.Orchestrator
                                          input: Optional[TInput] = None,
                                          instance_id: Optional[str] = None,
                                          start_at: Optional[datetime] = None,
-                                         reuse_id_policy: Optional[pb.OrchestrationIdReusePolicy] = None,
-                                         tags: Optional[dict[str, str]] = None,
-                                         version: Optional[str] = None) -> str:
+                                         reuse_id_policy: Optional[pb.OrchestrationIdReusePolicy] = None) -> str:
 
         name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
 
@@ -64,9 +60,8 @@ async def schedule_new_orchestration(self, orchestrator: Union[task.Orchestrator
             instanceId=instance_id if instance_id else uuid.uuid4().hex,
             input=wrappers_pb2.StringValue(value=shared.to_json(input)) if input is not None else None,
             scheduledStartTimestamp=helpers.new_timestamp(start_at) if start_at else None,
-            version=helpers.get_string_value(version if version else self.default_version),
+            version=helpers.get_string_value(None),
             orchestrationIdReusePolicy=reuse_id_policy,
-            tags=tags
         )
 
         self._logger.info(f"Starting new '{name}' instance with ID = '{req.instanceId}'.")
@@ -80,25 +75,30 @@ async def get_orchestration_state(self, instance_id: str, *, fetch_payloads: boo
 
     async def wait_for_orchestration_start(self, instance_id: str, *,
                                            fetch_payloads: bool = False,
-                                           timeout: int = 60) -> Optional[OrchestrationState]:
+                                           timeout: int = 0) -> Optional[OrchestrationState]:
         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.")
-            res: pb.GetInstanceResponse = await self._stub.WaitForInstanceStart(req, timeout=timeout)
+            grpc_timeout = None if timeout == 0 else timeout
+            self._logger.info(
+                f"Waiting {'indefinitely' if timeout == 0 else f'up to {timeout}s'} for instance '{instance_id}' to start.")
+            res: pb.GetInstanceResponse = await self._stub.WaitForInstanceStart(req, timeout=grpc_timeout)
             return new_orchestration_state(req.instanceId, res)
         except grpc.RpcError as rpc_error:
             if rpc_error.code() == grpc.StatusCode.DEADLINE_EXCEEDED:  # type: ignore
+                # Replace gRPC error with the built-in TimeoutError
                 raise TimeoutError("Timed-out waiting for the orchestration to start")
             else:
                 raise
 
     async def wait_for_orchestration_completion(self, instance_id: str, *,
                                                 fetch_payloads: bool = True,
-                                                timeout: int = 60) -> Optional[OrchestrationState]:
+                                                timeout: int = 0) -> Optional[OrchestrationState]:
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         try:
-            self._logger.info(f"Waiting {timeout}s for instance '{instance_id}' to complete.")
-            res: pb.GetInstanceResponse = await self._stub.WaitForInstanceCompletion(req, timeout=timeout)
+            grpc_timeout = None if timeout == 0 else timeout
+            self._logger.info(
+                f"Waiting {'indefinitely' if timeout == 0 else f'up to {timeout}s'} for instance '{instance_id}' to complete.")
+            res: pb.GetInstanceResponse = await self._stub.WaitForInstanceCompletion(req, timeout=grpc_timeout)
             state = new_orchestration_state(req.instanceId, res)
             if not state:
                 return None
@@ -114,6 +114,7 @@ async def wait_for_orchestration_completion(self, instance_id: str, *,
             return state
         except grpc.RpcError as rpc_error:
             if rpc_error.code() == grpc.StatusCode.DEADLINE_EXCEEDED:  # type: ignore
+                # Replace gRPC error with the built-in TimeoutError
                 raise TimeoutError("Timed-out waiting for the orchestration to complete")
             else:
                 raise

durabletask/task.py

@@ -283,6 +283,7 @@ def get_tasks(self) -> list[Task]:
     def on_child_completed(self, task: Task[T]):
         pass
 
+
 class WhenAllTask(CompositeTask[list[T]]):
     """A task that completes when all of its child tasks complete."""
 

durabletask/worker.py

@@ -880,13 +880,13 @@ class ExecutionResults:
     actions: list[pb.OrchestratorAction]
     encoded_custom_status: Optional[str]
 
-
     def __init__(
             self, actions: list[pb.OrchestratorAction], encoded_custom_status: Optional[str]
     ):
         self.actions = actions
         self.encoded_custom_status = encoded_custom_status
 
+
 class _OrchestrationExecutor:
     _generator: Optional[task.Orchestrator] = None
 

tests/durabletask/test_client.py

@@ -21,6 +21,7 @@ def test_get_grpc_channel_secure():
         get_grpc_channel(HOST_ADDRESS, True, interceptors=INTERCEPTORS)
         mock_channel.assert_called_once_with(HOST_ADDRESS, mock_credentials.return_value)
 
+
 def test_get_grpc_channel_default_host_address():
     with patch('grpc.insecure_channel') as mock_channel:
         get_grpc_channel(None, False, interceptors=INTERCEPTORS)

tests/durabletask/test_orchestration_e2e.py

@@ -316,7 +316,6 @@ def parent_orchestrator(ctx: task.OrchestrationContext, count: int):
             output = "Recursive termination = {recurse}"
             task_hub_client.terminate_orchestration(instance_id, output=output, recursive=recurse)
 
-
             metadata = task_hub_client.wait_for_orchestration_completion(instance_id, timeout=30)
 
             assert metadata is not None

tests/durabletask/test_orchestration_e2e_async.py

@@ -1,7 +1,6 @@
 import asyncio
 import json
 import threading
-import time
 from datetime import timedelta
 
 import pytest
@@ -16,13 +15,14 @@
 #       durabletask-go --port 4001
 pytestmark = [pytest.mark.e2e, pytest.mark.anyio]
 
+
 @pytest.fixture
 def anyio_backend():
     return 'asyncio'
 
 
 async def test_empty_orchestration():
- 
+
     invoked = False
 
     def empty_orchestrator(ctx: task.OrchestrationContext, _):
@@ -35,7 +35,7 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
         w.start()
 
         c = AsyncTaskHubGrpcClient()
-        id = await c.schedule_new_orchestration(empty_orchestrator, tags={'Tagged': 'true'})
+        id = await c.schedule_new_orchestration(empty_orchestrator)
         state = await c.wait_for_orchestration_completion(id, timeout=30)
         await c.aclose()
 
@@ -59,7 +59,7 @@ def sequence(ctx: task.OrchestrationContext, start_val: int):
         numbers = [start_val]
         current = start_val
         for _ in range(10):
-            current = yield ctx.call_activity(plus_one, input=current, tags={'Activity': 'PlusOne'})
+            current = yield ctx.call_activity(plus_one, input=current)
             numbers.append(current)
         return numbers
 
@@ -70,7 +70,7 @@ def sequence(ctx: task.OrchestrationContext, start_val: int):
         w.start()
 
         client = AsyncTaskHubGrpcClient()
-        id = await client.schedule_new_orchestration(sequence, input=1, tags={'Orchestration': 'Sequence'})
+        id = await client.schedule_new_orchestration(sequence, input=1)
         state = await client.wait_for_orchestration_completion(id, timeout=30)
         await client.aclose()
 
@@ -485,5 +485,3 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
     assert state.serialized_input is None
     assert state.serialized_output is None
     assert state.serialized_custom_status == "\"foobaz\""
-
-

tests/durabletask/test_orchestration_executor.py

@@ -634,7 +634,7 @@ def orchestrator(ctx: task.OrchestrationContext, _):
         yield ctx.call_sub_orchestrator(suborchestrator, input=None)
 
     registry = worker._Registry()
-    suborchestrator_name = registry.add_orchestrator(suborchestrator)
+    registry.add_orchestrator(suborchestrator)
     orchestrator_name = registry.add_orchestrator(orchestrator)
 
     exec_evt = helpers.new_execution_started_event(orchestrator_name, TEST_INSTANCE_ID, encoded_input=None)
@@ -666,7 +666,7 @@ def orchestrator(ctx: task.OrchestrationContext, _):
         yield ctx.call_sub_orchestrator(suborchestrator, input=None, app_id="target-app")
 
     registry = worker._Registry()
-    suborchestrator_name = registry.add_orchestrator(suborchestrator)
+    registry.add_orchestrator(suborchestrator)
     orchestrator_name = registry.add_orchestrator(orchestrator)
 
     exec_evt = helpers.new_execution_started_event(orchestrator_name, TEST_INSTANCE_ID, encoded_input=None)

tests/durabletask/test_orchestration_wait.py

@@ -1,11 +1,9 @@
-from unittest.mock import patch, ANY, Mock
+from unittest.mock import Mock
 
 from durabletask.client import TaskHubGrpcClient
-from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
-from durabletask.internal.shared import (get_default_host_address,
-                                         get_grpc_channel)
 import pytest
 
+
 @pytest.mark.parametrize("timeout", [None, 0, 5])
 def test_wait_for_orchestration_start_timeout(timeout):
     instance_id = "test-instance"
@@ -34,6 +32,7 @@ def test_wait_for_orchestration_start_timeout(timeout):
     else:
         assert kwargs.get('timeout') == timeout
 
+
 @pytest.mark.parametrize("timeout", [None, 0, 5])
 def test_wait_for_orchestration_completion_timeout(timeout):
     instance_id = "test-instance"