feat: Python error via grpc

pyproject.toml

@@ -47,6 +47,7 @@ ansys-units = "^0.3.3"
 docker = ">=7.1.0"
 grpcio = "^1.30.0"
 grpcio-health-checking = "^1.30.0"
+grpcio-status = "^1.30.0"
 lxml = ">=4.9.2"
 nltk = ">=3.9.1"
 numpy= ">=1.14.0,<3.0.0"

src/ansys/fluent/core/services/interceptors.py

@@ -1,12 +1,15 @@
 """Interceptor classes to use with gRPC services."""
 
+import builtins
 import logging
 import os
 from typing import Any
 
 from google.protobuf.json_format import MessageToDict
 from google.protobuf.message import Message
+from google.rpc import error_details_pb2
 import grpc
+from grpc_status import rpc_status
 
 from ansys.fluent.core.services.batch_ops import BatchOps
 
@@ -15,6 +18,10 @@
 truncate_len: int = log_bytes_limit // 5
 
 
+def _upper_snake_case_to_camel_case(name: str) -> str:
+    return "".join([word.capitalize() for word in name.split("_") if word])
+
+
 def _truncate_grpc_str(message: Message) -> str:
     message_bytes = message.ByteSize()
     message_str = str(MessageToDict(message))
@@ -107,7 +114,23 @@ def _intercept_call(
         response = continuation(client_call_details, request)
         if response.exception() is not None and response.code() != grpc.StatusCode.OK:
             ex = response.exception()
-            new_ex = RuntimeError(
+            new_ex_cls = RuntimeError
+            status = rpc_status.from_call(ex)
+            if status:
+                for detail in status.details:
+                    if detail.Is(error_details_pb2.ErrorInfo.DESCRIPTOR):
+                        info = error_details_pb2.ErrorInfo()
+                        detail.Unpack(info)
+                        if info.domain == "Python":
+                            reason = info.reason
+                            ex_cls_name = _upper_snake_case_to_camel_case(reason)
+                            if hasattr(builtins, ex_cls_name):
+                                cls = getattr(builtins, ex_cls_name)
+                                if issubclass(cls, Exception):
+                                    print(f"Found exception class: {cls}")
+                                    new_ex_cls = cls
+                                    break
+            new_ex = new_ex_cls(
                 ex.details() if isinstance(ex, grpc.RpcError) else str(ex)
             )
             new_ex.__context__ = ex

tests/test_error_handling.py

@@ -22,3 +22,12 @@ def test_fluent_fatal_error(error_code, raises, new_solver_session):
             # as these are mostly instant, exception should usually be raised on the second gRPC call
             scheme_eval("(pp 'fatal_error_testing)")
             time.sleep(0.1)
+
+
+def test_custom_python_error_via_grpc(datamodel_api_version_new, new_solver_session):
+    solver = new_solver_session
+    # This may need to be updated if the error type changes in the server
+    with pytest.raises(RuntimeError, match="prefereces not found!"):
+        solver._se_service.get_state("prefereces", "General")
+    with pytest.raises(ValueError, match="Datamodel rules for prefereces not found!"):
+        solver._se_service.get_specs("prefereces", "General")