python: Introduce a new connection API that is a bit less stateful.

python/dazl/__init__.py

@@ -50,6 +50,7 @@
     exercise_by_key,
 )
 from .ledger import Command
+from .ledger.grpc import connect
 from .pretty.table import write_acs
 from .prim import ContractData, ContractId, DazlError, FrozenDict as frozendict, Party
 from .util.logging import setup_default_logger

python/dazl/ledger/errors.py

@@ -0,0 +1,22 @@
+# Copyright (c) 2017-2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+__all__ = ["CallbackReturnWarning", "ProtocolWarning"]
+
+
+class CallbackReturnWarning(Warning):
+    """
+    Raised when a user callback on a stream returns a value. These objects have no meaning and are
+    ignored by dazl.
+
+    This warning is raised primarily because older versions of dazl interpreted returning commands
+    from a callback as a request to send commands to the underlying ledger, and this is not
+    supported in newer APIs.
+    """
+
+
+class ProtocolWarning(Warning):
+    """
+    Warnings that are raised when dazl detects incompatibilities between the Ledger API server-side
+    implementation and dazl.
+    """

python/dazl/ledger/grpc/__init__.py

@@ -0,0 +1,42 @@
+# Copyright (c) 2017-2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from os import PathLike
+from typing import Collection, Optional, Union, overload
+
+from ...prim import Party, TimeDeltaLike
+from ..config import Config
+from .conn_aio import Connection
+
+__all__ = ["connect", "Connection"]
+
+
+# TODO: Figure out clever ways to make this function's type signature easier to maintain while
+#  preserving its ease of use to callers.
+@overload
+def connect(
+    url: str,
+    *,
+    read_as: "Union[None, Party, Collection[Party]]" = None,
+    act_as: "Union[None, Party, Collection[Party]]" = None,
+    admin: "Optional[bool]" = False,
+    ledger_id: "Optional[str]" = None,
+    application_name: "Optional[str]" = None,
+    oauth_token: "Optional[str]" = None,
+    ca: "Optional[bytes]" = None,
+    ca_file: "Optional[PathLike]" = None,
+    cert: "Optional[bytes]" = None,
+    cert_file: "Optional[PathLike]" = None,
+    cert_key: "Optional[bytes]" = None,
+    cert_key_file: "Optional[PathLike]" = None,
+    connect_timeout: "Optional[TimeDeltaLike]" = None,
+    enable_http_proxy: "bool" = True,
+) -> Connection:
+    ...
+
+
+def connect(**kwargs):
+    """
+    Connect to a gRPC Ledger API implementation and return a connection that uses asyncio.
+    """
+    config = Config.create(**kwargs)
+    return Connection(config)

python/dazl/ledger/grpc/channel.py

@@ -0,0 +1,58 @@
+# Copyright (c) 2017-2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from grpc import (
+    AuthMetadataContext,
+    AuthMetadataPlugin,
+    AuthMetadataPluginCallback,
+    composite_channel_credentials,
+    metadata_call_credentials,
+    ssl_channel_credentials,
+)
+from grpc.aio import Channel, insecure_channel, secure_channel
+
+from ..config import Config
+
+__all__ = ["create_channel"]
+
+
+def create_channel(config: "Config") -> "Channel":
+    """
+    Create a :class:`Channel` for the specified configuration.
+    """
+    target = f"{config.url.host}:{config.url.port}"
+    options = [
+        ("grpc.max_send_message_length", -1),
+        ("grpc.max_receive_message_length", -1),
+    ]
+    if config.url.enable_http_proxy:
+        options.append(("grpc.enable_http_proxy", 0))
+
+    if (config.url.scheme in ("https", "grpcs")) or config.ssl:
+        credentials = ssl_channel_credentials(
+            root_certificates=config.ssl.ca,
+            private_key=config.ssl.cert_key,
+            certificate_chain=config.ssl.cert,
+        )
+        if config.access.token:
+            credentials = composite_channel_credentials(
+                credentials, metadata_call_credentials(GrpcAuth(config))
+            )
+        return secure_channel(target, credentials, options)
+    else:
+        return insecure_channel(target, options)
+
+
+class GrpcAuth(AuthMetadataPlugin):
+    def __init__(self, config: "Config"):
+        self._config = config
+
+    def __call__(self, context: "AuthMetadataContext", callback: "AuthMetadataPluginCallback"):
+        options = []
+
+        # TODO: Add support here for refresh tokens
+        token = self._config.access.token
+        if token:
+            options.append(("Authorization", "Bearer " + self._config.access.token))
+
+        callback(options, None)

python/dazl/ledger/grpc/codec_aio.py

@@ -11,7 +11,7 @@
 # References:
 #  * https://github.com/digital-asset/daml/blob/main/ledger-service/http-json/src/main/scala/com/digitalasset/http/CommandService.scala
 
-from typing import Any, List, Optional, Sequence, Tuple, Union
+from typing import TYPE_CHECKING, Any, List, Optional, Sequence, Tuple, Union
 
 from ..._gen.com.daml.ledger.api.v1.admin.party_management_service_pb2 import (
     PartyDetails as G_PartyDetails,