Adds support for Unsubscribe Configuration

dapr/clients/grpc/_response.py

@@ -15,7 +15,7 @@
 
 import threading
 from enum import Enum
-from typing import Dict, Optional, Union, Sequence
+from typing import Dict, Optional, Union, Sequence, List
 
 from google.protobuf.any_pb2 import Any as GrpcAny
 from google.protobuf.message import Message as GrpcMessage
@@ -34,6 +34,7 @@
 import json
 
 from dapr.proto import api_v1
+from dapr.proto import api_service_v1
 
 
 class DaprResponse:
@@ -671,23 +672,32 @@ def __init__(self):
     def get_items(self):
         return self.items
 
-    def watch_configuration(self, stub, store_name, keys, config_metadata):
+    def watch_configuration(self, stub: api_service_v1.DaprStub, store_name: str,
+                            keys: List[str], config_metadata: Optional[Dict[str, str]] = dict()):
         req = api_v1.SubscribeConfigurationRequest(
             store_name=store_name, keys=keys, metadata=config_metadata)
         thread = threading.Thread(target=self._read_subscribe_config, args=(stub, req))
         thread.daemon = True
         thread.start()
-
-    def _read_subscribe_config(self, stub, req):
-        responses = stub.SubscribeConfigurationAlpha1(req)
-        for response in responses:
-            for item in response.items:
-                self.items.append(
-                    ConfigurationItem(
-                        key=item.key,
-                        value=item.value,
-                        version=item.version,
-                        metadata=item.metadata))
+        self.keys = keys
+        self.store_name = store_name
+
+    def _read_subscribe_config(self, stub: api_service_v1.DaprStub,
+                               req: api_v1.SubscribeConfigurationRequest):
+        try:
+            responses = stub.SubscribeConfigurationAlpha1(req)
+            for response in responses:
+                for item in response.items:
+                    self.items.append(
+                        ConfigurationItem(
+                            key=item.key,
+                            value=item.value,
+                            version=item.version,
+                            metadata=item.metadata))
+        except Exception:
+            print(f"{self.store_name} configuration watcher for keys "
+                  f"{self.keys} stopped.")
+            pass
 
 
 class TopicEventResponseStatus(Enum):

dapr/clients/grpc/client.py

@@ -36,6 +36,7 @@
 from dapr.clients.grpc._state import StateOptions, StateItem
 from dapr.conf import settings
 from dapr.proto import api_v1, api_service_v1, common_v1
+from dapr.proto.runtime.v1.dapr_pb2 import UnsubscribeConfigurationResponse
 
 from dapr.clients.grpc._helpers import MetadataTuple, DaprClientInterceptor, to_bytes
 from dapr.clients.grpc._request import (
@@ -932,7 +933,7 @@ def get_configuration(
     async def subscribe_configuration(
             self,
             store_name: str,
-            keys: str,
+            keys: List[str],
             config_metadata: Optional[Dict[str, str]] = dict()) -> ConfigurationWatcher:
         """Gets changed value from a config store with a key
 
@@ -963,6 +964,26 @@ async def subscribe_configuration(
         configWatcher.watch_configuration(self._stub, store_name, keys, config_metadata)
         return configWatcher
 
+    def unsubscribe_configuration(
+            self,
+            store_name: str,
+            key: str) -> bool:
+        """Unsubscribes from configuration changes.
+
+            Args:
+                store_name (str): the state store name to unsubscribe from
+                key (str): the key of the key-value pair to unsubscribe from
+
+            Returns:
+                bool: True if unsubscribed successfully, False otherwise
+        """
+        warn('The Unsubscribe Configuration API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=key)
+        self._stub.UnsubscribeConfigurationAlpha1(req)
+        response: UnsubscribeConfigurationResponse = self._stub.UnsubscribeConfigurationAlpha1(req)
+        return response.ok
+
     def wait(self, timeout_s: float):
         """Waits for sidecar to be available within the timeout.
 

dapr/proto/runtime/v1/appcallback_pb2.pyi

@@ -256,6 +256,7 @@ class TopicSubscription(google.protobuf.message.Message):
     TOPIC_FIELD_NUMBER: builtins.int
     METADATA_FIELD_NUMBER: builtins.int
     ROUTES_FIELD_NUMBER: builtins.int
+    DEAD_LETTER_TOPIC_FIELD_NUMBER: builtins.int
     pubsub_name: typing.Text
     """Required. The name of the pubsub containing the topic below to subscribe to."""
 
@@ -272,15 +273,19 @@ class TopicSubscription(google.protobuf.message.Message):
         is still invoked but the matching path is sent in the TopicEventRequest.
         """
         pass
+    dead_letter_topic: typing.Text
+    """The optional dead letter queue for this topic to send events to."""
+
     def __init__(self,
         *,
         pubsub_name: typing.Text = ...,
         topic: typing.Text = ...,
         metadata: typing.Optional[typing.Mapping[typing.Text, typing.Text]] = ...,
         routes: typing.Optional[global___TopicRoutes] = ...,
+        dead_letter_topic: typing.Text = ...,
         ) -> None: ...
     def HasField(self, field_name: typing_extensions.Literal["routes",b"routes"]) -> builtins.bool: ...
-    def ClearField(self, field_name: typing_extensions.Literal["metadata",b"metadata","pubsub_name",b"pubsub_name","routes",b"routes","topic",b"topic"]) -> None: ...
+    def ClearField(self, field_name: typing_extensions.Literal["dead_letter_topic",b"dead_letter_topic","metadata",b"metadata","pubsub_name",b"pubsub_name","routes",b"routes","topic",b"topic"]) -> None: ...
 global___TopicSubscription = TopicSubscription
 
 class TopicRoutes(google.protobuf.message.Message):