Merge pull request #260 from digital-asset/python-type-libs

python: Add some type libraries

Author: da-tanabe

VERSION

@@ -1 +1 @@
-7.5.4
+7.5.5

poetry.lock

@@ -3,7 +3,7 @@
 
 [tool.poetry]
 name = "dazl"
-version = "7.5.4"
+version = "7.5.5"
 description = "high-level Ledger API client for DAML ledgers"
 license = "Apache-2.0"
 authors = ["Davin K. Tanabe <davin.tanabe@digitalasset.com>"]
@@ -18,6 +18,7 @@ python = "^3.6"
 aiohttp = { version = "*", optional = true }
 dataclasses = { version = "*", python = "~=3.6.0" }
 google-auth = { version = "*", optional = true }
+googleapis_common_protos = "^1"
 grpcio = ">=1.32.0"
 oauthlib = { version = "*", optional = true }
 prometheus_client = { version = "*", optional = true }
@@ -42,6 +43,7 @@ sphinx = "^4"
 sphinx-autobuild = "*"
 sphinx-markdown-builder = "^0.5.4"
 sphinxcontrib-trio = "^1.1.2"
+types-protobuf = "*"
 watchdog = "*"
 
 [tool.poetry.extras]

pyproject.toml

@@ -1,7 +1,8 @@
 # Copyright (c) 2017-2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
-
-from typing import TYPE_CHECKING, Callable, TypeVar
+import asyncio
+import concurrent.futures
+from typing import TYPE_CHECKING, Callable, Optional, TypeVar
 
 if TYPE_CHECKING:
     from . import Connection
@@ -15,6 +16,7 @@
     "ConnectionClosedError",
     "_rewrite_exceptions",
     "_translate_exceptions",
+    "_allow_cancel",
 ]
 
 
@@ -95,9 +97,46 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
             raise ConnectionClosedError() from exc_val
 
 
+class AllowCancellation:
+    def __init__(self, allow: Callable[[], bool]):
+        self.allow = allow
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> Optional[bool]:
+        if exc_val is not None:
+            # These two exceptions are actually the same type under the hood--however, nothing
+            # that I can find in the Python docs suggest that this HAS to be the case. Either
+            # way, suppress cancellation errors that happen when we are closed, because .
+            return self.allow() and (
+                exc_type is asyncio.CancelledError or exc_type is concurrent.futures.CancelledError
+            )
+
+        return None
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> Optional[bool]:
+        if exc_val is not None:
+            # These two exceptions are actually the same type under the hood--however, nothing
+            # that I can find in the Python docs suggest that this HAS to be the case. Either
+            # way, suppress cancellation errors that happen when we are closed, because .
+            return self.allow() and (
+                exc_type is asyncio.CancelledError or exc_type is concurrent.futures.CancelledError
+            )
+
+        return None
+
+
 def _translate_exceptions(conn: "Connection") -> ExceptionTranslator:
     """
     Return an (async) context manager that translates exceptions thrown from low-level gRPC calls
     to high-level dazl exceptions.
     """
     return ExceptionTranslator(conn)
+
+
+def _allow_cancel(allow: Callable[[], bool]) -> AllowCancellation:
+    return AllowCancellation(allow)

python/dazl/ledger/errors.py

@@ -63,7 +63,7 @@
 from ..api_types import ArchiveEvent, Boundary, Command, CreateEvent, ExerciseResponse, PartyInfo
 from ..config import Config
 from ..config.access import PropertyBasedAccessConfig
-from ..errors import ProtocolWarning, _translate_exceptions
+from ..errors import ProtocolWarning, _allow_cancel, _translate_exceptions
 from .channel import create_channel
 from .codec_aio import Codec
 
@@ -644,8 +644,16 @@ def __init__(
         self._filters = filters
         self._offset_range = offset_range
         self._response_stream = None  # type: Optional[UnaryStreamCall]
+        self._closed = False
+
+    @property
+    def is_closed(self) -> bool:
+        return self._closed
 
     async def close(self) -> None:
+        # make sure to mark the object as "closed"; when we're closed, we don't mind cancellation
+        # errors, because we're the one triggering the cancellation
+        self._closed = True
         if self._response_stream is not None:
             self._response_stream.cancel()
             self._response_stream = None
@@ -670,7 +678,7 @@ async def items(self):
             In this case, the first returned object is a :class:`Boundary` with ``offset=None``.
         """
         log = self.conn.config.logger
-        async with _translate_exceptions(self.conn), self:
+        async with _translate_exceptions(self.conn), self, _allow_cancel(lambda: self._closed):
             filters = await self.conn.codec.encode_filters(self._filters)
             filters_by_party = {party: filters for party in self.conn.config.access.read_as}
             tx_filter_pb = G_TransactionFilter(filters_by_party=filters_by_party)

python/dazl/ledger/grpc/conn_aio.py

@@ -1,6 +1,6 @@
 # Copyright (c) 2017-2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
-
+import asyncio
 from asyncio import CancelledError, Future, InvalidStateError, ensure_future, get_event_loop, sleep
 from collections.abc import Mapping as MappingBase
 import sys
@@ -205,6 +205,9 @@ async def _main(self) -> None:
                 if snapshot is not None:
                     self._snapshot = snapshot
                     self._snapshot_fut.set_result(snapshot)
+        except CancelledError:
+            if self._snapshot_fut is not None and not self._snapshot_fut.done():
+                self._snapshot_fut.cancel()
         except Exception as ex:
             # No one is actually waiting for our result, so if we don't move our exception to
             # an awaitable that people actually care about, it will never be seen!

python/dazl/ledgerutil/acs.py

@@ -53,3 +53,12 @@ async def accept_roles(sandbox, party):
         async with conn.query("Main:InviteReceiverRole") as query:
             async for event in query.creates():
                 await conn.exercise(event.contract_id, "AcceptInviteReceiverRole")
+
+
+@pytest.mark.asyncio
+async def test_acs_can_async_read(sandbox):
+    async with connect_with_new_party(url=sandbox, dar=PostOffice) as p:
+        async with ACS(p.connection, {"Main:PostmanRole"}) as acs:
+            await acs.read()
+
+    assert True