Modernize PR-new code for Python 3.10+ baseline

Apply three Python 3.10+ idioms to code introduced by this PR:

- PEP 604 union syntax: replace Optional[X] with X | None in grpc_options.py and grpc_resiliency.py.

- Dataclass tightening: add slots=True, kw_only=True to FailureTracker, GrpcRetryPolicyOptions, GrpcChannelOptions, GrpcWorkerResiliencyOptions, and GrpcClientResiliencyOptions. Update the two positional FailureTracker(...) call sites in client.py to use threshold=... kwargs.

- PEP 617 parenthesized context managers: rewrite the 10 chained 'with patch(...), patch(...):' blocks added by this PR in test_client.py. Pre-existing chained sites are left untouched to keep the diff surgical.

Internal-only change (no public API or behavior impact). 85/85 resiliency-focused tests pass; 232/8 passed/skipped in the broader non-e2e suite.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: Bernd Verst

durabletask/client.py

@@ -215,7 +215,7 @@ def __init__(self, *,
         # ``close()`` wait deterministically for an in-flight recreate.
         self._recreate_done_event = threading.Event()
         self._client_failure_tracker = FailureTracker(
-            self._resiliency_options.channel_recreate_failure_threshold
+            threshold=self._resiliency_options.channel_recreate_failure_threshold,
         )
         self._resiliency_interceptor = ClientResiliencyInterceptor(
             self._client_failure_tracker,
@@ -705,7 +705,7 @@ def __init__(self, *,
         # ``close()`` await an in-flight recreate deterministically.
         self._recreate_done_event = asyncio.Event()
         self._client_failure_tracker = FailureTracker(
-            self._resiliency_options.channel_recreate_failure_threshold
+            threshold=self._resiliency_options.channel_recreate_failure_threshold,
         )
         self._resiliency_interceptor = AsyncClientResiliencyInterceptor(
             self._client_failure_tracker,

durabletask/grpc_options.py

@@ -5,10 +5,10 @@
 
 from dataclasses import dataclass, field
 import json
-from typing import Any, Optional
+from typing import Any
 
 
-@dataclass
+@dataclass(slots=True, kw_only=True)
 class GrpcRetryPolicyOptions:
     """Configuration for transport-level gRPC retries."""
 
@@ -64,16 +64,16 @@ def to_service_config(self) -> dict[str, Any]:
         }
 
 
-@dataclass
+@dataclass(slots=True, kw_only=True)
 class GrpcChannelOptions:
     """Configuration for transport-level gRPC channel behavior."""
 
-    max_receive_message_length: Optional[int] = None
-    max_send_message_length: Optional[int] = None
-    keepalive_time_ms: Optional[int] = None
-    keepalive_timeout_ms: Optional[int] = None
-    keepalive_permit_without_calls: Optional[bool] = None
-    retry_policy: Optional[GrpcRetryPolicyOptions] = None
+    max_receive_message_length: int | None = None
+    max_send_message_length: int | None = None
+    keepalive_time_ms: int | None = None
+    keepalive_timeout_ms: int | None = None
+    keepalive_permit_without_calls: bool | None = None
+    retry_policy: GrpcRetryPolicyOptions | None = None
     raw_options: list[tuple[str, Any]] = field(default_factory=list)
 
     def to_grpc_options(self) -> list[tuple[str, Any]]:
@@ -102,7 +102,7 @@ def to_grpc_options(self) -> list[tuple[str, Any]]:
         return options
 
 
-@dataclass
+@dataclass(slots=True, kw_only=True)
 class GrpcWorkerResiliencyOptions:
     """Configuration for worker-side gRPC resiliency behavior."""
 
@@ -129,7 +129,7 @@ def __post_init__(self) -> None:
             )
 
 
-@dataclass
+@dataclass(slots=True, kw_only=True)
 class GrpcClientResiliencyOptions:
     """Configuration for client-side gRPC resiliency behavior."""
 

durabletask/internal/grpc_resiliency.py

@@ -4,7 +4,7 @@
 import random
 import threading
 from dataclasses import dataclass, field
-from typing import Callable, Optional
+from typing import Callable
 
 import grpc
 import grpc.aio
@@ -35,7 +35,7 @@ def get_full_jitter_delay_seconds(
     return random.random() * upper_bound
 
 
-@dataclass
+@dataclass(slots=True, kw_only=True)
 class FailureTracker:
     """Counts consecutive transport failures with thread-safe mutation.
 
@@ -114,7 +114,7 @@ def intercept_unary_unary(self, continuation, client_call_details, request):
         self._record_outcome(client_call_details.method, error)
         return response
 
-    def _record_outcome(self, method: str, error: Optional[BaseException]) -> None:
+    def _record_outcome(self, method: str, error: BaseException | None) -> None:
         if error is None:
             self._failure_tracker.record_success()
             return
@@ -162,7 +162,7 @@ async def intercept_unary_unary(self, continuation, client_call_details, request
         self._record_outcome(client_call_details.method, None)
         return response
 
-    def _record_outcome(self, method: str, error: Optional[BaseException]) -> None:
+    def _record_outcome(self, method: str, error: BaseException | None) -> None:
         if error is None:
             self._failure_tracker.record_success()
             return

tests/durabletask/test_client.py

@@ -509,8 +509,9 @@ def test_client_stores_resiliency_options_for_recreation():
     resiliency = GrpcClientResiliencyOptions(channel_recreate_failure_threshold=7)
     channel_options = GrpcChannelOptions(max_receive_message_length=1234)
     interceptors = [DefaultClientInterceptorImpl(METADATA)]
-    with patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()), patch(
-            "durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=MagicMock()
+    with (
+        patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()),
+        patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=MagicMock()),
     ):
         client = TaskHubGrpcClient(
             host_address="localhost:4001",
@@ -542,12 +543,17 @@ def test_sync_client_recreates_sdk_owned_channel_with_original_transport_inputs(
 
     timer = MagicMock()
 
-    with patch(
+    with (
+        patch(
             "durabletask.client.shared.get_grpc_channel",
             side_effect=[first_channel, second_channel],
-    ) as mock_get_channel, patch(
-            "durabletask.client.stubs.TaskHubSidecarServiceStub", side_effect=[first_stub, second_stub]
-    ), patch("threading.Timer", return_value=timer) as mock_timer:
+        ) as mock_get_channel,
+        patch(
+            "durabletask.client.stubs.TaskHubSidecarServiceStub",
+            side_effect=[first_stub, second_stub],
+        ),
+        patch("threading.Timer", return_value=timer) as mock_timer,
+    ):
         client = TaskHubGrpcClient(
             host_address=host_address,
             secure_channel=True,
@@ -596,12 +602,17 @@ def test_sync_client_close_closes_retired_channels_immediately():
     second_stub.GetInstance.return_value = MagicMock(exists=False)
     close_timer = MagicMock(name="close-timer")
 
-    with patch(
+    with (
+        patch(
             "durabletask.client.shared.get_grpc_channel",
             side_effect=[first_channel, second_channel],
-    ), patch(
-            "durabletask.client.stubs.TaskHubSidecarServiceStub", side_effect=[first_stub, second_stub]
-    ), patch("threading.Timer", return_value=close_timer):
+        ),
+        patch(
+            "durabletask.client.stubs.TaskHubSidecarServiceStub",
+            side_effect=[first_stub, second_stub],
+        ),
+        patch("threading.Timer", return_value=close_timer),
+    ):
         client = TaskHubGrpcClient(
             resiliency_options=GrpcClientResiliencyOptions(
                 channel_recreate_failure_threshold=1,
@@ -632,13 +643,17 @@ def test_sync_client_close_closes_all_retired_sdk_channels_immediately():
     timer1 = MagicMock(name="close-timer-1")
     timer2 = MagicMock(name="close-timer-2")
 
-    with patch(
+    with (
+        patch(
             "durabletask.client.shared.get_grpc_channel",
             side_effect=[first_channel, second_channel, third_channel],
-    ), patch(
+        ),
+        patch(
             "durabletask.client.stubs.TaskHubSidecarServiceStub",
             side_effect=[first_stub, second_stub, third_stub],
-    ), patch("threading.Timer", side_effect=[timer1, timer2]):
+        ),
+        patch("threading.Timer", side_effect=[timer1, timer2]),
+    ):
         client = TaskHubGrpcClient(
             resiliency_options=GrpcClientResiliencyOptions(
                 channel_recreate_failure_threshold=1,
@@ -681,8 +696,9 @@ def test_sync_client_resets_failure_tracking_after_long_poll_deadline(
     stub.GetInstance.side_effect = FakeRpcError(grpc.StatusCode.UNAVAILABLE)
     getattr(stub, stub_method_name).side_effect = FakeRpcError(grpc.StatusCode.DEADLINE_EXCEEDED)
 
-    with patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()), patch(
-            "durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub
+    with (
+        patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()),
+        patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub),
     ):
         client = TaskHubGrpcClient(
             resiliency_options=GrpcClientResiliencyOptions(channel_recreate_failure_threshold=2)
@@ -700,9 +716,13 @@ def test_sync_client_does_not_recreate_caller_owned_channel():
     stub = MagicMock()
     stub.GetInstance.side_effect = FakeRpcError(grpc.StatusCode.UNAVAILABLE)
 
-    with patch("durabletask.client.shared.get_grpc_channel") as mock_get_channel, patch(
+    with (
+        patch("durabletask.client.shared.get_grpc_channel") as mock_get_channel,
+        patch(
             "durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub
-    ) as mock_stub, patch("threading.Timer") as mock_timer:
+        ) as mock_stub,
+        patch("threading.Timer") as mock_timer,
+    ):
         client = TaskHubGrpcClient(
             channel=provided_channel,
             resiliency_options=GrpcClientResiliencyOptions(channel_recreate_failure_threshold=1),
@@ -739,15 +759,20 @@ def test_sync_client_recreate_cooldown_prevents_immediate_repeated_recreation():
     timer1 = MagicMock(name="close-timer-1")
     timer2 = MagicMock(name="close-timer-2")
 
-    with patch(
+    with (
+        patch(
             "durabletask.client.shared.get_grpc_channel",
             side_effect=[first_channel, second_channel, third_channel],
-    ) as mock_get_channel, patch(
+        ) as mock_get_channel,
+        patch(
             "durabletask.client.stubs.TaskHubSidecarServiceStub",
             side_effect=[first_stub, second_stub, third_stub],
-    ), patch(
+        ),
+        patch(
             "durabletask.client.time.monotonic", side_effect=[100.0, 101.0, 131.0]
-    ), patch("threading.Timer", side_effect=[timer1, timer2]) as mock_timer:
+        ),
+        patch("threading.Timer", side_effect=[timer1, timer2]) as mock_timer,
+    ):
         client = TaskHubGrpcClient(
             host_address=HOST_ADDRESS,
             resiliency_options=GrpcClientResiliencyOptions(
@@ -811,8 +836,9 @@ def test_sync_client_resets_failure_tracking_after_success():
         MagicMock(exists=False),
     ]
 
-    with patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()), patch(
-            "durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub
+    with (
+        patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()),
+        patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub),
     ):
         client = TaskHubGrpcClient(
             resiliency_options=GrpcClientResiliencyOptions(channel_recreate_failure_threshold=2)
@@ -831,8 +857,9 @@ def test_sync_client_resets_failure_tracking_after_application_error():
         FakeRpcError(grpc.StatusCode.INVALID_ARGUMENT),
     ]
 
-    with patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()), patch(
-            "durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub
+    with (
+        patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()),
+        patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub),
     ):
         client = TaskHubGrpcClient(
             resiliency_options=GrpcClientResiliencyOptions(channel_recreate_failure_threshold=2)
@@ -906,8 +933,9 @@ async def test_async_client_does_not_count_wait_for_orchestration_deadline():
     stub.GetInstance = AsyncMock(side_effect=make_aio_rpc_error(grpc.StatusCode.UNAVAILABLE))
     stub.WaitForInstanceCompletion = AsyncMock(side_effect=make_aio_rpc_error(grpc.StatusCode.DEADLINE_EXCEEDED))
 
-    with patch("durabletask.client.shared.get_async_grpc_channel", return_value=MagicMock()), patch(
-            "durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub
+    with (
+        patch("durabletask.client.shared.get_async_grpc_channel", return_value=MagicMock()),
+        patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub),
     ):
         client = AsyncTaskHubGrpcClient(
             resiliency_options=GrpcClientResiliencyOptions(channel_recreate_failure_threshold=2)
@@ -939,15 +967,20 @@ async def blocked_close_retired_channel(self, channel):
         await release_cleanup.wait()
         await channel.close()
 
-    with patch(
+    with (
+        patch(
             "durabletask.client.shared.get_async_grpc_channel",
             side_effect=[first_channel, second_channel],
-    ), patch(
-            "durabletask.client.stubs.TaskHubSidecarServiceStub", side_effect=[first_stub, second_stub]
-    ), patch.object(
+        ),
+        patch(
+            "durabletask.client.stubs.TaskHubSidecarServiceStub",
+            side_effect=[first_stub, second_stub],
+        ),
+        patch.object(
             AsyncTaskHubGrpcClient,
             "_close_retired_channel",
             new=blocked_close_retired_channel,
+        ),
     ):
         client = AsyncTaskHubGrpcClient(
             resiliency_options=GrpcClientResiliencyOptions(