Error handling (#659)

* Improves error handling for grpc Status messages Signed-off-by: Elena Kolevska <[email protected]> * Missed a line Signed-off-by: Elena Kolevska <[email protected]> * Adds deps Signed-off-by: Elena Kolevska <[email protected]> * Add type ignores Signed-off-by: Elena Kolevska <[email protected]> * Handles rich errors for get and delete state Signed-off-by: Elena Kolevska <[email protected]> * Update all tests for state endpoints Signed-off-by: Elena Kolevska <[email protected]> * Adds error handling for pubsub publishing Signed-off-by: Elena Kolevska <[email protected]> * Tests cover all status details Signed-off-by: Elena Kolevska <[email protected]> * Linter fixes Signed-off-by: Elena Kolevska <[email protected]> * Adds an error-handling example and client docs Signed-off-by: Elena Kolevska <[email protected]> * Runs ruff Signed-off-by: Elena Kolevska <[email protected]> * Adds json() Signed-off-by: Elena Kolevska <[email protected]> * Removes unneeded import Signed-off-by: Elena Kolevska <[email protected]> * Type fix Signed-off-by: Elena Kolevska <[email protected]> * linter Signed-off-by: Elena Kolevska <[email protected]> * Apply suggestions from code review Signed-off-by: Bernd Verst <[email protected]> * example fix Signed-off-by: Elena Kolevska <[email protected]> * Removing link which does not exist yet Signed-off-by: Bernd Verst <[email protected]> --------- Signed-off-by: Elena Kolevska <[email protected]> Signed-off-by: Bernd Verst <[email protected]> Co-authored-by: Bernd Verst <[email protected]>
dapr · Feb 6, 2024 · fc0e9d1 · fc0e9d1
1 parent ef73209
commit fc0e9d1
Show file tree

Hide file tree

Showing 10 changed files with 586 additions and 24 deletions.
diff --git a/dapr/clients/exceptions.py b/dapr/clients/exceptions.py
@@ -12,9 +12,14 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
-
+import json
 from typing import Optional
 
+from google.protobuf.json_format import MessageToDict
+from grpc import RpcError  # type: ignore
+from grpc_status import rpc_status  # type: ignore
+from google.rpc import error_details_pb2  # type: ignore
+
 ERROR_CODE_UNKNOWN = 'UNKNOWN'
 ERROR_CODE_DOES_NOT_EXIST = 'ERR_DOES_NOT_EXIST'
 
@@ -38,3 +43,92 @@ def as_dict(self):
             'errorCode': self._error_code,
             'raw_response_bytes': self._raw_response_bytes,
         }
+
+
+class StatusDetails:
+    def __init__(self):
+        self.error_info = None
+        self.retry_info = None
+        self.debug_info = None
+        self.quota_failure = None
+        self.precondition_failure = None
+        self.bad_request = None
+        self.request_info = None
+        self.resource_info = None
+        self.help = None
+        self.localized_message = None
+
+    def as_dict(self):
+        return {attr: getattr(self, attr) for attr in self.__dict__}
+
+
+class DaprGrpcError(RpcError):
+    def __init__(self, err: RpcError):
+        self._status_code = err.code()
+        self._err_message = err.details()
+        self._details = StatusDetails()
+
+        self._grpc_status = rpc_status.from_call(err)
+        self._parse_details()
+
+    def _parse_details(self):
+        if self._grpc_status is None:
+            return
+
+        for detail in self._grpc_status.details:
+            if detail.Is(error_details_pb2.ErrorInfo.DESCRIPTOR):
+                self._details.error_info = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.RetryInfo.DESCRIPTOR):
+                self._details.retry_info = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.DebugInfo.DESCRIPTOR):
+                self._details.debug_info = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.QuotaFailure.DESCRIPTOR):
+                self._details.quota_failure = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.PreconditionFailure.DESCRIPTOR):
+                self._details.precondition_failure = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.BadRequest.DESCRIPTOR):
+                self._details.bad_request = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.RequestInfo.DESCRIPTOR):
+                self._details.request_info = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.ResourceInfo.DESCRIPTOR):
+                self._details.resource_info = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.Help.DESCRIPTOR):
+                self._details.help = serialize_status_detail(detail)
+            elif detail.Is(error_details_pb2.LocalizedMessage.DESCRIPTOR):
+                self._details.localized_message = serialize_status_detail(detail)
+
+    def code(self):
+        return self._status_code
+
+    def details(self):
+        """
+        We're keeping the method name details() so it matches the grpc.RpcError interface.
+        @return:
+        """
+        return self._err_message
+
+    def error_code(self):
+        if not self.status_details() or not self.status_details().error_info:
+            return ERROR_CODE_UNKNOWN
+        return self.status_details().error_info.get('reason', ERROR_CODE_UNKNOWN)
+
+    def status_details(self):
+        return self._details
+
+    def get_grpc_status(self):
+        return self._grpc_status
+
+    def json(self):
+        error_details = {
+            'status_code': self.code().name,
+            'message': self.details(),
+            'error_code': self.error_code(),
+            'details': self._details.as_dict(),
+        }
+        return json.dumps(error_details)
+
+
+def serialize_status_detail(status_detail):
+    if not status_detail:
+        return None
+    return MessageToDict(status_detail, preserving_proto_field_name=True)
diff --git a/dapr/clients/grpc/client.py b/dapr/clients/grpc/client.py
@@ -37,7 +37,7 @@
     RpcError,
 )
 
-from dapr.clients.exceptions import DaprInternalError
+from dapr.clients.exceptions import DaprInternalError, DaprGrpcError
 from dapr.clients.grpc._state import StateOptions, StateItem
 from dapr.clients.grpc._helpers import getWorkflowRuntimeStatus
 from dapr.conf import settings
@@ -451,8 +451,11 @@ def publish_event(
             metadata=publish_metadata,
         )
 
-        # response is google.protobuf.Empty
-        _, call = self._stub.PublishEvent.with_call(req, metadata=metadata)
+        try:
+            # response is google.protobuf.Empty
+            _, call = self._stub.PublishEvent.with_call(req, metadata=metadata)
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
 
         return DaprResponse(call.initial_metadata())
 
@@ -496,10 +499,13 @@ def get_state(
         if not store_name or len(store_name) == 0 or len(store_name.strip()) == 0:
             raise ValueError('State store name cannot be empty')
         req = api_v1.GetStateRequest(store_name=store_name, key=key, metadata=state_metadata)
-        response, call = self._stub.GetState.with_call(req, metadata=metadata)
-        return StateResponse(
-            data=response.data, etag=response.etag, headers=call.initial_metadata()
-        )
+        try:
+            response, call = self._stub.GetState.with_call(req, metadata=metadata)
+            return StateResponse(
+                data=response.data, etag=response.etag, headers=call.initial_metadata()
+            )
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
 
     def get_bulk_state(
         self,
@@ -545,7 +551,11 @@ def get_bulk_state(
         req = api_v1.GetBulkStateRequest(
             store_name=store_name, keys=keys, parallelism=parallelism, metadata=states_metadata
         )
-        response, call = self._stub.GetBulkState.with_call(req, metadata=metadata)
+
+        try:
+            response, call = self._stub.GetBulkState.with_call(req, metadata=metadata)
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
 
         items = []
         for item in response.items:
@@ -603,7 +613,11 @@ def query_state(
         if not store_name or len(store_name) == 0 or len(store_name.strip()) == 0:
             raise ValueError('State store name cannot be empty')
         req = api_v1.QueryStateRequest(store_name=store_name, query=query, metadata=states_metadata)
-        response, call = self._stub.QueryStateAlpha1.with_call(req)
+
+        try:
+            response, call = self._stub.QueryStateAlpha1.with_call(req)
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
 
         results = []
         for item in response.results:
@@ -692,8 +706,11 @@ def save_state(
         )
 
         req = api_v1.SaveStateRequest(store_name=store_name, states=[state])
-        _, call = self._stub.SaveState.with_call(req, metadata=metadata)
-        return DaprResponse(headers=call.initial_metadata())
+        try:
+            _, call = self._stub.SaveState.with_call(req, metadata=metadata)
+            return DaprResponse(headers=call.initial_metadata())
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
 
     def save_bulk_state(
         self, store_name: str, states: List[StateItem], metadata: Optional[MetadataTuple] = None
@@ -749,8 +766,12 @@ def save_bulk_state(
         ]
 
         req = api_v1.SaveStateRequest(store_name=store_name, states=req_states)
-        _, call = self._stub.SaveState.with_call(req, metadata=metadata)
-        return DaprResponse(headers=call.initial_metadata())
+
+        try:
+            _, call = self._stub.SaveState.with_call(req, metadata=metadata)
+            return DaprResponse(headers=call.initial_metadata())
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
 
     def execute_state_transaction(
         self,
@@ -814,8 +835,12 @@ def execute_state_transaction(
         req = api_v1.ExecuteStateTransactionRequest(
             storeName=store_name, operations=req_ops, metadata=transactional_metadata
         )
-        _, call = self._stub.ExecuteStateTransaction.with_call(req, metadata=metadata)
-        return DaprResponse(headers=call.initial_metadata())
+
+        try:
+            _, call = self._stub.ExecuteStateTransaction.with_call(req, metadata=metadata)
+            return DaprResponse(headers=call.initial_metadata())
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
 
     def delete_state(
         self,
@@ -878,8 +903,12 @@ def delete_state(
             options=state_options,
             metadata=state_metadata,
         )
-        _, call = self._stub.DeleteState.with_call(req, metadata=metadata)
-        return DaprResponse(headers=call.initial_metadata())
+
+        try:
+            _, call = self._stub.DeleteState.with_call(req, metadata=metadata)
+            return DaprResponse(headers=call.initial_metadata())
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
 
     def get_secret(
         self,
@@ -1524,7 +1553,11 @@ def get_metadata(self) -> GetMetadataResponse:
         information about supported features in the form of component
         capabilities.
         """
-        _resp, call = self._stub.GetMetadata.with_call(GrpcEmpty())
+        try:
+            _resp, call = self._stub.GetMetadata.with_call(GrpcEmpty())
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
+
         response: api_v1.GetMetadataResponse = _resp  # type alias
         # Convert to more pythonic formats
         active_actors_count = {

diff --git a/daprdocs/content/en/python-sdk-docs/python-client.md b/daprdocs/content/en/python-sdk-docs/python-client.md
@@ -74,6 +74,26 @@ If your Dapr instance is configured to require the `DAPR_API_TOKEN` environment
 set it in the environment and the client will use it automatically.  
 You can read more about Dapr API token authentication [here](https://docs.dapr.io/operations/security/api-token/).
 
+## Error handling
+Initially, errors in Dapr followed the [Standard gRPC error model](https://grpc.io/docs/guides/error/#standard-error-model). However, to provide more detailed and informative error messages, in version 1.13 an enhanced error model has been introduced which aligns with the gRPC [Richer error model](https://grpc.io/docs/guides/error/#richer-error-model). In response, the Python SDK implemented `DaprGrpcError`, a custom exception class designed to improve the developer experience.  
+It's important to note that the transition to using `DaprGrpcError` for all gRPC status exceptions is a work in progress. As of now, not every API call in the SDK has been updated to leverage this custom exception. We are actively working on this enhancement and welcome contributions from the community.
+
+Example of handling `DaprGrpcError` exceptions when using the Dapr python-SDK:
+
+```python
+try:
+    d.save_state(store_name=storeName, key=key, value=value)
+except DaprGrpcError as err:
+    print(f'Status code: {err.code()}')
+    print(f"Message: {err.message()}")
+    print(f"Error code: {err.error_code()}")
+    print(f"Error info(reason): {err.error_info.reason}")
+    print(f"Resource info (resource type): {err.resource_info.resource_type}")
+    print(f"Resource info (resource name): {err.resource_info.resource_name}")
+    print(f"Bad request (field): {err.bad_request.field_violations[0].field}")
+    print(f"Bad request (description): {err.bad_request.field_violations[0].description}")
+```
+
 
 ## Building blocks
 

diff --git a/examples/error_handling/README.md b/examples/error_handling/README.md
@@ -0,0 +1,57 @@
+# Example - Error handling
+
+This guide demonstrates handling `DaprGrpcError` errors when using the Dapr python-SDK. It's important to note that not all Dapr gRPC status errors are currently captured and transformed into a `DaprGrpcError` by the SDK. Efforts are ongoing to enhance this aspect, and contributions are welcome. For detailed information on error handling in Dapr, refer to the [official documentation](https://docs.dapr.io/reference/).
+
+The example involves creating a DaprClient and invoking the save_state method. 
+It uses the default configuration from Dapr init in [self-hosted mode](https://github.com/dapr/cli#install-dapr-on-your-local-machine-self-hosted). 
+
+## Pre-requisites
+
+- [Dapr CLI and initialized environment](https://docs.dapr.io/getting-started)
+- [Install Python 3.8+](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 used:
+
+<!-- STEP
+name: Run error handling example
+expected_stdout_lines:
+- "== APP == Status code: StatusCode.INVALID_ARGUMENT"
+- "== APP == Message: input key/keyPrefix 'key||' can't contain '||'"
+- "== APP == Error code: DAPR_STATE_ILLEGAL_KEY"
+- "== APP == Error info(reason): DAPR_STATE_ILLEGAL_KEY"
+- "== APP == Resource info (resource type): state"
+- "== APP == Resource info (resource name): statestore"
+- "== APP == Bad request (field): key||"
+- "== APP == Bad request (description): input key/keyPrefix 'key||' can't contain '||'"
+- "== APP == JSON: {\"status_code\": \"INVALID_ARGUMENT\", \"message\": \"input key/keyPrefix 'key||' can't contain '||'\", \"error_code\": \"DAPR_STATE_ILLEGAL_KEY\", \"details\": {\"error_info\": {\"@type\": \"type.googleapis.com/google.rpc.ErrorInfo\", \"reason\": \"DAPR_STATE_ILLEGAL_KEY\", \"domain\": \"dapr.io\"}, \"retry_info\": null, \"debug_info\": null, \"quota_failure\": null, \"precondition_failure\": null, \"bad_request\": {\"@type\": \"type.googleapis.com/google.rpc.BadRequest\", \"field_violations\": [{\"field\": \"key||\", \"description\": \"input key/keyPrefix 'key||' can't contain '||'\"}]}, \"request_info\": null, \"resource_info\": {\"@type\": \"type.googleapis.com/google.rpc.ResourceInfo\", \"resource_type\": \"state\", \"resource_name\": \"statestore\"}, \"help\": null, \"localized_message\": null}}"
+timeout_seconds: 5
+-->
+
+```bash
+dapr run -- python3 error_handling.py
+```
+<!-- END_STEP -->
+
+The output should be as follows:
+
+```
+== APP == Status code: INVALID_ARGUMENT
+== APP == Message: input key/keyPrefix 'key||' can't contain '||'
+== APP == Error code: DAPR_STATE_ILLEGAL_KEY
+== APP == Error info(reason): DAPR_STATE_ILLEGAL_KEY
+== APP == Resource info (resource type): state
+== APP == Resource info (resource name): statestore
+== APP == Bad request (field): key||
+== APP == Bad request (description): input key/keyPrefix 'key||' can't contain '||'
+== APP == JSON: {"status_code": "INVALID_ARGUMENT", "message": "input key/keyPrefix 'key||' can't contain '||'", "error_code": "DAPR_STATE_ILLEGAL_KEY", "details": {"error_info": {"@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": "DAPR_STATE_ILLEGAL_KEY", "domain": "dapr.io"}, "retry_info": null, "debug_info": null, "quota_failure": null, "precondition_failure": null, "bad_request": {"@type": "type.googleapis.com/google.rpc.BadRequest", "field_violations": [{"field": "key||", "description": "input key/keyPrefix 'key||' can't contain '||'"}]}, "request_info": null, "resource_info": {"@type": "type.googleapis.com/google.rpc.ResourceInfo", "resource_type": "state", "resource_name": "statestore"}, "help": null, "localized_message": null}}
+```
diff --git a/examples/error_handling/error_handling.py b/examples/error_handling/error_handling.py
@@ -0,0 +1,41 @@
+from dapr.clients import DaprClient
+from dapr.clients.exceptions import DaprGrpcError
+
+with DaprClient() as d:
+    storeName = 'statestore'
+
+    key = 'key||'
+    value = 'value_1'
+
+    # Wait for sidecar to be up within 5 seconds.
+    d.wait(5)
+
+    # Save single state.
+    try:
+        d.save_state(store_name=storeName, key=key, value=value)
+    except DaprGrpcError as err:
+        print(f'Status code: {err.code()}', flush=True)
+        print(f'Message: {err.details()}', flush=True)
+        print(f'Error code: {err.error_code()}', flush=True)
+
+        if err.status_details().error_info is not None:
+            print(f'Error info(reason): {err.status_details().error_info["reason"]}', flush=True)
+        if err.status_details().resource_info is not None:
+            print(
+                f'Resource info (resource type): {err.status_details().resource_info["resource_type"]}',
+                flush=True,
+            )
+            print(
+                f'Resource info (resource name): {err.status_details().resource_info["resource_name"]}',
+                flush=True,
+            )
+        if err.status_details().bad_request is not None:
+            print(
+                f'Bad request (field): {err.status_details().bad_request["field_violations"][0]["field"]}',
+                flush=True,
+            )
+            print(
+                f'Bad request (description): {err.status_details().bad_request["field_violations"][0]["description"]}',
+                flush=True,
+            )
+        print(f'JSON: {err.json()}', flush=True)
diff --git a/setup.cfg b/setup.cfg
@@ -27,6 +27,7 @@ zip_safe = False
 install_requires =
     protobuf >= 4.22
     grpcio >= 1.37.0
+    grpcio-status>=1.37.0
     aiohttp >= 3.9.0b0
     python-dateutil >= 2.8.1
     typing-extensions>=4.4.0