Added python methods for Distributed Lock API (#431)

* Initial implementation for distributed lock API.

Adds initial API support and unit tests.

Missing:
* DocStrings and documentation support
* Context Manager-friendly wrapper or API.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Doc strings and converts ProtBuff enum to python native.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* try_lock returns TryLockResponse objects.

gRPC's client `try_lock()` method now returns an TryLockResponse
object that can be used both:
* on boolean context (if-clause) directly, keeping behaviour from
  previous commits, and
* as a ContextManager (with-clause), which would ease auto-unlock
  and make this implementation similar to DotNet's one.

Updated tests to account for new functionality.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Updates exemples to acount for new Distributed Locks API.

This should also work as an e2e test.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Reponse objects inherit from DaprResponse.

* Updated the example's output to be deterministic.
* Updated example's README.md to account for the new output.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Updates to address PR comments.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Addressing PR comments (cont'ed.)

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Removing dead test code.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Fixes "Alpha version" msg and adds error message to `assert` clause.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Fix type annotation for dapr client and remove deprecated field.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Fixes some lint issues.

Seems that the lint configuration/version used by the
validation steps in GitHub are not matching the ones
available in the devcontainer, as GitHub is reporting
lint errors that are not being returned by tox
validation tasks.

Regardless, fixing the issues reported to get the
PR approved.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Register distributed lock mechanical markdown tests.

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Move input validation to _helpers, fix mention of "Alpha version".

Signed-off-by: Tiago Alves Macambira <[email protected]>

* Fix typo in last commit.

Signed-off-by: Tiago Alves Macambira <[email protected]>

Author: tmacam

dapr/clients/grpc/_helpers.py

@@ -14,7 +14,7 @@
 """
 
 from collections import namedtuple
-from typing import Dict, List, Union, Tuple
+from typing import Dict, List, Union, Tuple, Optional
 
 from google.protobuf.any_pb2 import Any as GrpcAny
 from google.protobuf.message import Message as GrpcMessage
@@ -66,7 +66,7 @@ def to_bytes(data: Union[str, bytes]) -> bytes:
     elif isinstance(data, str):
         return data.encode('utf-8')
     else:
-        raise(f'invalid data type {type(data)}')
+        raise f'invalid data type {type(data)}'
 
 
 def to_str(data: Union[str, bytes]) -> str:
@@ -76,7 +76,7 @@ def to_str(data: Union[str, bytes]) -> str:
     elif isinstance(data, bytes):
         return data.decode('utf-8')
     else:
-        raise(f'invalid data type {type(data)}')
+        raise f'invalid data type {type(data)}'
 
 
 class _ClientCallDetails(
@@ -166,3 +166,12 @@ def intercept_unary_unary(
         # Call continuation
         response = continuation(new_call_details, request)
         return response
+
+
+# Data validation helpers
+
+
+def validateNotBlankString(**kwargs: Optional[str]):
+    for field_name, value in kwargs.items():
+        if not value or not value.strip():
+            raise ValueError(f"{field_name} name cannot be empty or blank")

dapr/clients/grpc/_request.py

@@ -289,7 +289,8 @@ def __init__(
             data: Union[bytes, str],
             etag: Optional[str] = None,
             operation_type: TransactionOperationType = TransactionOperationType.upsert):
-        """Initializes TransactionalStateOperation item from :obj:`runtime_v1.TransactionalStateOperation`.
+        """Initializes TransactionalStateOperation item from
+        :obj:`runtime_v1.TransactionalStateOperation`.
 
         Args:
             key (str): state's key.

dapr/clients/grpc/_response.py

@@ -13,9 +13,12 @@
 limitations under the License.
 """
 
+from __future__ import annotations
+
+import contextlib
 import threading
 from enum import Enum
-from typing import Dict, Optional, Union, Sequence, List
+from typing import Dict, Optional, Union, Sequence, List, TYPE_CHECKING
 
 from google.protobuf.any_pb2 import Any as GrpcAny
 from google.protobuf.message import Message as GrpcMessage
@@ -36,6 +39,11 @@
 from dapr.proto import api_v1
 from dapr.proto import api_service_v1
 
+# Avoid circular import dependency by only importing DaprGrpcClient
+# for type checking
+if TYPE_CHECKING:
+    from dapr.clients.grpc.client import DaprGrpcClient
+
 
 class DaprResponse:
     """A base class for Dapr Response.
@@ -744,3 +752,102 @@ def __init__(
     def status(self) -> TopicEventResponseStatus:
         """Gets the status."""
         return self._status
+
+
+class UnlockResponseStatus(Enum):
+    success = api_v1.UnlockResponse.Status.SUCCESS
+    '''The Unlock operation for the referred lock was successful.'''
+
+    lock_does_not_exist = api_v1.UnlockResponse.Status.LOCK_UNEXIST
+    ''''The unlock operation failed: the referred lock does not exist.'''
+
+    lock_belongs_to_others = api_v1.UnlockResponse.Status.LOCK_BELONG_TO_OTHERS
+    '''The unlock operation failed: the referred lock belongs to another owner.'''
+
+    internal_error = api_v1.UnlockResponse.Status.INTERNAL_ERROR
+    '''An internal error happened while handling the Unlock operation'''
+
+
+class UnlockResponse(DaprResponse):
+    '''The response of an unlock operation.
+
+    This inherits from DaprResponse
+
+    Attributes:
+        status (UnlockResponseStatus): the status of the unlock operation.
+    '''
+
+    def __init__(
+        self,
+        status: UnlockResponseStatus,
+        headers: MetadataTuple = (),
+    ):
+        """Initializes a UnlockResponse.
+
+        Args:
+            status (UnlockResponseStatus): The status of the response.
+            headers (Tuple, optional): the headers from Dapr gRPC response.
+        """
+        super().__init__(headers)
+        self._status = status
+
+    @property
+    def status(self) -> UnlockResponseStatus:
+        """Gets the status."""
+        return self._status
+
+
+class TryLockResponse(contextlib.AbstractContextManager, DaprResponse):
+    '''The response of a try_lock operation.
+
+    This inherits from DaprResponse and AbstractContextManager.
+
+    Attributes:
+        success (bool): the result of the try_lock operation.
+    '''
+    def __init__(
+        self,
+        success: bool,
+        client: DaprGrpcClient,
+        store_name: str,
+        resource_id: str,
+        lock_owner: str,
+        headers: MetadataTuple = (),
+    ):
+        """Initializes a TryLockResponse.
+
+        Args:
+            success (bool): the result of the try_lock operation.
+            client (DaprClient): a reference to the dapr client used for the TryLock request.
+            store_name (str): the lock store name used in the TryLock request.
+            resource_id (str): the lock key or identifier used in the TryLock request.
+            lock_owner (str):  the lock owner identifier used in the TryLock request.
+            headers (Tuple, optional): the headers from Dapr gRPC response.
+        """
+        super().__init__(headers)
+        self._success = success
+        self._client = client
+        self._store_name = store_name
+        self._resource_id = resource_id
+        self._lock_owner = lock_owner
+
+    def __bool__(self) -> bool:
+        return self._success
+
+    @property
+    def success(self) -> bool:
+        """Gets the response success status."""
+        return self._success
+
+    def __exit__(self, *exc) -> None:
+        ''''Automatically unlocks the lock if this TryLockResponse was used as
+        a ContextManager / `with` statement.
+
+        Notice: we are not checking the result of the unlock operation.
+        If this is something  you care about it might be wiser creating
+        your own ContextManager that logs or otherwise raises exceptions
+        if unlock doesn't return `UnlockResponseStatus.success`.
+        '''
+        if self._success:
+            self._client.unlock(self._store_name, self._resource_id, self._lock_owner)
+        # else: there is no point unlocking a lock we did not acquire.

dapr/clients/grpc/client.py

@@ -38,7 +38,12 @@
 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._helpers import (
+    DaprClientInterceptor,
+    MetadataTuple,
+    to_bytes,
+    validateNotBlankString,
+)
 from dapr.clients.grpc._request import (
     InvokeMethodRequest,
     BindingRequest,
@@ -50,14 +55,17 @@
     GetSecretResponse,
     GetBulkSecretResponse,
     InvokeMethodResponse,
+    UnlockResponseStatus,
     StateResponse,
     BulkStatesResponse,
     BulkStateItem,
     ConfigurationResponse,
     ConfigurationItem,
     QueryResponse,
     QueryResponseItem,
-    ConfigurationWatcher
+    ConfigurationWatcher,
+    TryLockResponse,
+    UnlockResponse,
 )
 
 
@@ -984,6 +992,102 @@ def unsubscribe_configuration(
         response: UnsubscribeConfigurationResponse = self._stub.UnsubscribeConfigurationAlpha1(req)
         return response.ok
 
+    def try_lock(
+            self,
+            store_name: str,
+            resource_id: str,
+            lock_owner: str,
+            expiry_in_seconds: int) -> TryLockResponse:
+        """Tries to get a lock with an expiry.
+
+            You can use the result of this operation directly on an `if` statement:
+
+                if client.try_lock(store_name, resource_id, first_client_id, expiry_s):
+                    # lock acquired successfully...
+
+            You can also inspect the response's `success` attribute:
+
+                    response = client.try_lock(store_name, resource_id, first_client_id, expiry_s)
+                    if response.success:
+                        # lock acquired successfully...
+
+            Finally, you can use this response with a `with` statement, and have the lock
+            be automatically unlocked after the with-statement scope ends
+
+                with client.try_lock(store_name, resource_id, first_client_id, expiry_s) as lock:
+                    if lock:
+                        # lock acquired successfully...
+                # Lock automatically unlocked at this point, no need to call client->unlock(...)
+
+            Args:
+                store_name (str): the lock store name, e.g. `redis`.
+                resource_id (str): the lock key. e.g. `order_id_111`.
+                                    It stands for "which resource I want to protect".
+                lock_owner (str):  indicates the identifier of lock owner.
+                expiry_in_seconds (int): The length of time (in seconds) for which this lock
+                    will be held and after which it expires.
+
+            Returns:
+                :class:`TryLockResponse`: With the result of the try-lock operation.
+        """
+        # Warnings and input validation
+        warn('The Distributed Lock API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(store_name=store_name,
+                               resource_id=resource_id,
+                               lock_owner=lock_owner)
+        if not expiry_in_seconds or expiry_in_seconds < 1:
+            raise ValueError("expiry_in_seconds must be a positive number")
+        # Actual tryLock invocation
+        req = api_v1.TryLockRequest(
+            store_name=store_name,
+            resource_id=resource_id,
+            lock_owner=lock_owner,
+            expiryInSeconds=expiry_in_seconds)
+        response, call = self._stub.TryLockAlpha1.with_call(req)
+        return TryLockResponse(
+            success=response.success,
+            client=self,
+            store_name=store_name,
+            resource_id=resource_id,
+            lock_owner=lock_owner,
+            headers=call.initial_metadata())
+
+    def unlock(
+            self,
+            store_name: str,
+            resource_id: str,
+            lock_owner: str) -> UnlockResponse:
+        """Unlocks a lock.
+
+            Args:
+                store_name (str): the lock store name, e.g. `redis`.
+                resource_id (str): the lock key. e.g. `order_id_111`.
+                                    It stands for "which resource I want to protect".
+                lock_owner (str):  indicates the identifier of lock owner.
+                metadata (tuple, optional, DEPRECATED): gRPC custom metadata
+
+            Returns:
+                :class:`UnlockResponseStatus`: Status of the request,
+                    `UnlockResponseStatus.success` if it was successful of some other
+                    status otherwise.
+        """
+        # Warnings and input validation
+        warn('The Distributed Lock API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(store_name=store_name,
+                               resource_id=resource_id,
+                               lock_owner=lock_owner)
+        # Actual unlocking invocation
+        req = api_v1.UnlockRequest(
+            store_name=store_name,
+            resource_id=resource_id,
+            lock_owner=lock_owner)
+        response, call = self._stub.UnlockAlpha1.with_call(req)
+
+        return UnlockResponse(status=UnlockResponseStatus(response.status),
+                              headers=call.initial_metadata())
+
     def wait(self, timeout_s: float):
         """Waits for sidecar to be available within the timeout.
 

examples/README.md

@@ -12,6 +12,7 @@ These examples demonstrate how to use the Dapr Python SDK:
 | [Virtual actors](./demo_actor) | Try Dapr virtual actor features
 | [Secrets](./secret_store) | Get secrets from a defined secret store
 | [Distributed tracing](./w3c-tracing) | Leverage Dapr's built-in tracing support
+| [Distributed lock](./distributed_lock) | Keep your application safe from race conditions by using distributed locks
 
 ## More information
 

examples/distributed_lock/README.md

@@ -0,0 +1,63 @@
+# Example - Acquire and release distributed locks
+
+This example demonstrates the [Distributed Lock component] APIs in Dapr.
+It demonstrates the following APIs:
+- **try_lock**: Attempts to acquire a distributed lock from the lock store.
+- **unlock**: Attempts to release (a previously acquired) distributed lock
+
+It creates a client using `DaprClient`, uses a local lock store defined in
+[`./components/lockstore.yaml`](./components/lockstore.yaml) and invokes
+all the distributed lock API methods available as example.
+
+## Pre-requisites
+
+- [Dapr CLI and initialized environment](https://docs.dapr.io/getting-started)
+- [Install Python 3.7+](https://www.python.org/downloads/)
+
+## Install Dapr python-SDK
+
+<!-- Our CI/CD pipeline automatically installs the correct version, so we can skip this step in the automation -->
+
+```bash
+pip3 install dapr dapr-ext-grpc
+```
+
+## Run the example
+
+To run this example, the following code can be utilized:
+
+<!-- STEP
+name: Run state store example
+expected_stdout_lines:
+  - "== APP == Will try to acquire a lock from lock store named [lockstore]"
+  - "== APP == The lock is for a resource named [example-lock-resource]"
+  - "== APP == The client identifier is [example-client-id]"
+  - "== APP == The lock will will expire in 60 seconds."
+  - "== APP == Lock acquired successfully!!!"
+  - "== APP == We already released the lock so unlocking will not work."
+  - "== APP == We tried to unlock it anyway and got back [UnlockResponseStatus.lock_does_not_exist]"
+timeout_seconds: 5
+-->
+
+```bash
+dapr run --app-id=locksapp --app-protocol grpc --components-path components/ python3 lock.py
+```
+<!-- END_STEP -->
+
+The output should be as follows:
+
+```
+== APP == Will try to acquire a lock from lock store named [lockstore]
+== APP == The lock is for a resource named [example-lock-resource]
+== APP == The client identifier is [example-client-id]
+== APP == The lock will will expire in 60 seconds.
+== APP == Lock acquired successfully!!!
+== APP == We already released the lock so unlocking will not work.
+== APP == We tried to unlock it anyway and got back [UnlockResponseStatus.lock_does_not_exist]
+```
+
+## Error Handling
+
+The Dapr python-sdk will pass through errors that it receives from the Dapr runtime.
+
+[Distributed Lock component]: https://docs.dapr.io/developing-applications/building-blocks/distributed-lock/