Merge branch 'master' into renovate/pylint-3.x-lockfile

pyzeebe/channel/camunda_cloud_channel.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Optional
+from typing import Optional
 
 import grpc
 from oauthlib import oauth2
@@ -10,14 +10,15 @@
     InvalidCamundaCloudCredentialsError,
     InvalidOAuthCredentialsError,
 )
+from pyzeebe.types import ChannelArgumentType
 
 
 def create_camunda_cloud_channel(
     client_id: str,
     client_secret: str,
     cluster_id: str,
     region: str = "bru-2",
-    channel_options: Optional[Dict[str, Any]] = None,
+    channel_options: Optional[ChannelArgumentType] = None,
 ) -> grpc.aio.Channel:
     """
     Create channel connected to a Camunda Cloud cluster
@@ -27,7 +28,8 @@ def create_camunda_cloud_channel(
         client_secret (str): The client secret provided by Camunda Cloud
         cluster_id (str): The zeebe cluster id to connect to
         region (str): The cluster's region. Defaults to bru-2
-        channel_options (Optional[Dict], optional): GRPC channel options. See https://grpc.github.io/grpc/python/glossary.html
+        channel_options (Optional[Dict], optional): GRPC channel options.
+            See https://grpc.github.io/grpc/python/glossary.html
 
     Returns:
         grpc.aio.Channel: A GRPC Channel connected to the Zeebe gateway.
@@ -40,7 +42,7 @@ def create_camunda_cloud_channel(
     return grpc.aio.secure_channel(
         f"{cluster_id}.{region}.zeebe.camunda.io:443",
         channel_credentials,
-        options=get_channel_options(channel_options or {}),
+        options=get_channel_options(channel_options),
     )
 
 

pyzeebe/channel/channel_options.py

@@ -8,27 +8,34 @@
 https://docs.camunda.io/docs/product-manuals/zeebe/deployment-guide/operations/setting-up-a-cluster/#keep-alive-intervals
 """
 
-from typing import Any, Dict, Optional, Tuple
+from typing import Optional
 
-GRPC_CHANNEL_OPTIONS = {"grpc.keepalive_time_ms": 45_000}
+from pyzeebe.types import ChannelArgumentType
 
+GRPC_CHANNEL_OPTIONS_DEFAULT: ChannelArgumentType = (("grpc.keepalive_time_ms", 45_000),)
 
-def get_channel_options(options: Optional[Dict[str, Any]] = None) -> Tuple[Tuple[str, Any], ...]:
+
+def get_channel_options(options: Optional[ChannelArgumentType] = None) -> ChannelArgumentType:
     """
-    Convert options dict to tuple in expected format for creating the gRPC channel
+    Get default channel options for creating the gRPC channel.
 
     Args:
-        options (Dict[str, Any]): A key/value representation of `gRPC channel arguments_`.
+        options (Optional[ChannelArgumentType]): A tuple of gRPC channel arguments tuple.
+            e.g. (("grpc.keepalive_time_ms", 45_000),)
             Default: None (will use library defaults)
+            See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
 
     Returns:
-        Tuple[Tuple[str, Any], ...]: Options for the gRPC channel
-
-    .. _gRPC channel arguments:
-        https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
+        ChannelArgumentType: Options for the gRPC channel
     """
-    if options:
-        options = {**GRPC_CHANNEL_OPTIONS, **options}
-    else:
-        options = GRPC_CHANNEL_OPTIONS
-    return tuple((k, v) for k, v in options.items())
+    if options is not None:
+        existing = set()
+        _options = []
+
+        for a, b in (*options, *GRPC_CHANNEL_OPTIONS_DEFAULT):
+            if a not in existing:  # NOTE: Remove duplicates, fist one wins
+                existing.add(a)
+                _options.append((a, b))
+
+        return tuple(_options)  # (*options, GRPC_CHANNEL_OPTIONS)
+    return GRPC_CHANNEL_OPTIONS_DEFAULT

pyzeebe/channel/insecure_channel.py

@@ -1,21 +1,23 @@
-from typing import Any, Dict, Optional
+from typing import Optional
 
 import grpc
 
 from pyzeebe.channel.channel_options import get_channel_options
 from pyzeebe.channel.utils import create_address
+from pyzeebe.types import ChannelArgumentType
 
 
 def create_insecure_channel(
-    hostname: Optional[str] = None, port: Optional[int] = None, channel_options: Optional[Dict[str, Any]] = None
+    hostname: Optional[str] = None, port: Optional[int] = None, channel_options: Optional[ChannelArgumentType] = None
 ) -> grpc.aio.Channel:
     """
     Create an insecure channel
 
     Args:
         hostname (Optional[str], optional): Zeebe gateway hostname
         port (Optional[int], optional): Zeebe gateway port
-        channel_options (Optional[Dict], optional): GRPC channel options. See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
+        channel_options (Optional[Dict], optional): GRPC channel options.
+            See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
 
     Returns:
         grpc.aio.Channel: A GRPC Channel connected to the Zeebe gateway.

pyzeebe/channel/secure_channel.py

@@ -1,15 +1,16 @@
-from typing import Any, Dict, Optional
+from typing import Optional
 
 import grpc
 
 from pyzeebe.channel.channel_options import get_channel_options
 from pyzeebe.channel.utils import create_address
+from pyzeebe.types import ChannelArgumentType
 
 
 def create_secure_channel(
     hostname: Optional[str] = None,
     port: Optional[int] = None,
-    channel_options: Optional[Dict[str, Any]] = None,
+    channel_options: Optional[ChannelArgumentType] = None,
     channel_credentials: Optional[grpc.ChannelCredentials] = None,
 ) -> grpc.aio.Channel:
     """
@@ -18,8 +19,10 @@ def create_secure_channel(
     Args:
         hostname (Optional[str], optional): Zeebe gateway hostname
         port (Optional[int], optional): Zeebe gateway port
-        channel_options (Optional[Dict], optional): GRPC channel options. See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
-        channel_credentials (Optional[grpc.ChannelCredentials]): Channel credentials to use. Will use grpc.ssl_channel_credentials() if not provided.
+        channel_options (Optional[Dict], optional): GRPC channel options.
+            See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
+        channel_credentials (Optional[grpc.ChannelCredentials]): Channel credentials to use.
+            Will use grpc.ssl_channel_credentials() if not provided.
 
     Returns:
         grpc.aio.Channel: A GRPC Channel connected to the Zeebe gateway.

pyzeebe/types.py

@@ -1,6 +1,8 @@
-from typing import Any, Mapping
+from typing import Any, Mapping, Sequence, Tuple
 
 from typing_extensions import TypeAlias
 
 Headers: TypeAlias = Mapping[str, Any]
 Variables: TypeAlias = Mapping[str, Any]
+
+ChannelArgumentType: TypeAlias = Sequence[Tuple[str, Any]]

tests/unit/channel/channel_options_test.py

@@ -2,23 +2,23 @@
 
 
 def test_get_channel_options_returns_tuple_of_tuple_with_options():
-    assert get_channel_options() == (("grpc.keepalive_time_ms", 45000),)
+    assert get_channel_options() == (("grpc.keepalive_time_ms", 45_000),)
 
 
 def test_overrides_default_values_if_provided():
-    grpc_options = {"grpc.keepalive_time_ms": 4000}
+    grpc_options = (("grpc.keepalive_time_ms", 4_000),)
 
-    assert get_channel_options(grpc_options) == (("grpc.keepalive_time_ms", 4000),)
+    assert get_channel_options(grpc_options) == (("grpc.keepalive_time_ms", 4_000),)
 
 
 def test_adds_custom_options():
-    grpc_options = {
-        "grpc.keepalive_timeout_ms": 120000,
-        "grpc.http2.min_time_between_pings_ms": 60000,
-    }
+    grpc_options = (
+        ("grpc.keepalive_timeout_ms", 120_000),
+        ("grpc.http2.min_time_between_pings_ms", 60_000),
+    )
 
     assert get_channel_options(grpc_options) == (
-        ("grpc.keepalive_time_ms", 45000),
-        ("grpc.keepalive_timeout_ms", 120000),
-        ("grpc.http2.min_time_between_pings_ms", 60000),
+        ("grpc.keepalive_timeout_ms", 120_000),
+        ("grpc.http2.min_time_between_pings_ms", 60_000),
+        ("grpc.keepalive_time_ms", 45_000),
     )