python: Fix a bug where aborting early from a stream would close a connection. (#545)

Author: da-tanabe

VERSION

@@ -1 +1 @@
-8.7.0
+8.7.1

pyproject.toml

@@ -3,7 +3,7 @@
 
 [tool.poetry]
 name = "dazl"
-version = "8.7.0"
+version = "8.7.1"
 description = "high-level Ledger API client for Daml ledgers"
 license = "Apache-2.0"
 authors = ["Davin K. Tanabe <[email protected]>"]

python/dazl/__init__.py

@@ -57,4 +57,4 @@
     pass
 
 
-__version__ = "8.7.0"
+__version__ = "8.7.1"

python/dazl/ledger/errors.py

@@ -88,13 +88,19 @@ def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
+        if isinstance(exc_val, GeneratorExit):
+            return None
+
         if exc_val is not None and self.conn.is_closed:
             raise ConnectionClosedError() from exc_val
 
     async def __aenter__(self):
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
+        if isinstance(exc_val, GeneratorExit):
+            return None
+
         if exc_val is not None and self.conn.is_closed:
             raise ConnectionClosedError() from exc_val
 

python/tests/unit/test_ledger_early_stream_abort.py

@@ -0,0 +1,45 @@
+# Copyright (c) 2017-2025 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+import random
+from typing import Any
+
+from dazl import Party, connect
+from dazl.ledger.aio import Connection
+from dazl.testing import SandboxLauncher, sandbox
+import pytest
+
+from .dars import Simple
+
+
+@pytest.mark.asyncio
+async def test_early_stream_abort(sandbox: SandboxLauncher) -> None:
+    async with connect(url=sandbox.url) as conn:
+        await conn.upload_package(Simple.read_bytes())
+        alice = await conn.allocate_party(identifier_hint="Alice")
+        for i in range(100):
+            await conn.create(
+                "Simple:OperatorNotification",
+                {
+                    "operator": alice.party,
+                    "theObservers": [],
+                    "text": f"{random.random()}",
+                },
+                act_as=alice.party,
+            )
+
+        for i in range(10):
+            payload = await find_one(conn, alice.party, str(i / 10))
+
+
+async def find_one(conn: Connection, party: Party, prefix: str) -> Any:
+    async with conn.stream("Simple:OperatorNotification", read_as=party) as stream:
+        async for ev in stream.creates():
+            if ev.payload["text"].startswith(prefix):
+                return ev.payload
+
+    return None