Stopped using ExceptionIteratingRequestsError so that CancelledError thrown into async persistent subscription request iterators can be raised directly (this only happens since Python 3.12), also changed in sync client for behavioural consistency.

Author: johnbywater

kurrentdbclient/exceptions.py

@@ -2,6 +2,8 @@
 
 from typing import TYPE_CHECKING
 
+from typing_extensions import deprecated
+
 if TYPE_CHECKING:
     from typing import Any
     from uuid import UUID
@@ -246,6 +248,7 @@ class ReadOnlyReplicaNotFoundError(DiscoveryFailedError):
     """
 
 
+@deprecated("Actual error raised instead")
 class ExceptionIteratingRequestsError(KurrentDBClientError):
     """
     Raised when a persistent subscription errors whilst iterating requests.

kurrentdbclient/persistent.py

@@ -54,7 +54,6 @@
 from kurrentdbclient.events import RecordedEvent
 from kurrentdbclient.exceptions import (
     CancelledByClientError,
-    ExceptionIteratingRequestsError,
     KurrentDBClientError,
     NodeIsNotLeaderError,
     ProgrammingError,
@@ -620,7 +619,7 @@ async def stop(self, *, wait_until_stopped: bool = True) -> None:
             except grpc.RpcError as e:
                 raise handle_rpc_error(e) from e
             if self._read_reqs.errored:
-                raise ExceptionIteratingRequestsError from self._read_reqs.errored
+                raise self._read_reqs.errored
 
     async def ack(self, item: UUID | RecordedEvent) -> None:
         await self._read_reqs.ack(event_id=self._get_event_id(item))
@@ -638,6 +637,13 @@ def _get_event_id(item: UUID | RecordedEvent) -> UUID:
             return item.ack_id
         return item
 
+    def __del__(self) -> None:
+        # Safety net, last chance to cancel the streaming call.
+        if hasattr(self, "_stream_stream_call"):
+            self._stream_stream_call.cancel()
+        else:  # pragma: no cover
+            pass
+
 
 class PersistentSubscription(
     GrpcStreamer, AbstractPersistentSubscription, BasePersistentSubscription
@@ -717,7 +723,7 @@ def _get_next_read_resp(self) -> persistent_pb2.ReadResp:
                 and "Exception iterating requests!" in details
                 and self._read_reqs.errored
             ):
-                raise ExceptionIteratingRequestsError from self._read_reqs.errored
+                raise self._read_reqs.errored from None
             raise handle_rpc_error(e) from None
         assert isinstance(read_resp, persistent_pb2.ReadResp)
         return read_resp

kurrentdbclient/streams.py

@@ -239,6 +239,13 @@ async def stop(self, *, wait_until_stopped: bool = True) -> None:
             except grpc.RpcError as e:
                 raise handle_streams_rpc_error(e) from e
 
+    def __del__(self) -> None:
+        # Safety net, last chance to cancel the streaming call.
+        if hasattr(self, "_unary_stream_call"):
+            self._unary_stream_call.cancel()
+        else:  # pragma: no cover
+            pass
+
 
 class AsyncCatchupSubscription(AsyncReadResponse, AbstractAsyncCatchupSubscription):
     def __init__(

tests/test_client.py

@@ -50,7 +50,6 @@
     ConsumerTooSlowError,
     DeadlineExceededError,
     DiscoveryFailedError,
-    ExceptionIteratingRequestsError,
     ExceptionThrownByHandlerError,
     FailedPreconditionError,
     FollowerNotFoundError,
@@ -3564,10 +3563,10 @@ class NotUUID:
             pass
 
         # Ack with wrong type of object.
-        with self.assertRaises(ExceptionIteratingRequestsError) as cm:
+        with self.assertRaises(ValueError) as cm:
             for _ in subscription:
                 subscription.ack(NotUUID())  # type: ignore
-        self.assertIsInstance(cm.exception.__cause__, ValueError)
+        self.assertIn("is not a UUID", str(cm.exception))
 
     def test_subscription_to_stream_replay_parked(self) -> None:
         self.construct_client()

tests/test_client_async.py

@@ -35,7 +35,6 @@
     AlreadyExistsError,
     DeadlineExceededError,
     DiscoveryFailedError,
-    ExceptionIteratingRequestsError,
     ExceptionThrownByHandlerError,
     FollowerNotFoundError,
     GrpcDeadlineExceededError,
@@ -1601,9 +1600,10 @@ async def test_persistent_subscription_to_all(self) -> None:
 
         # Read subscription - error iterating requests is propagated.
         persistent_subscription = await self.client.read_subscription_to_all(group_name)
-        with self.assertRaises(ExceptionIteratingRequestsError):
+        with self.assertRaises(ValueError) as cm:
             async for _ in persistent_subscription:
                 await persistent_subscription.ack("a")  # type: ignore[arg-type]
+        self.assertIn("event_id 'a' is not a UUID", str(cm.exception))
 
         # Read subscription - success.
         persistent_subscription = await self.client.read_subscription_to_all(group_name)
@@ -2420,9 +2420,10 @@ async def test_persistent_subscription_to_stream(self) -> None:
         subscription = await self.client.read_subscription_to_stream(
             group_name, stream_name1
         )
-        with self.assertRaises(ExceptionIteratingRequestsError):
+        with self.assertRaises(ValueError) as cm:
             async for _ in subscription:
                 await subscription.ack("a")  # type: ignore[arg-type]
+        self.assertIn("event_id 'a' is not a UUID", str(cm.exception))
 
         # Read subscription - success.
         subscription = await self.client.read_subscription_to_stream(