feat: Python error via grpc

pyproject.toml

@@ -27,6 +27,7 @@ dependencies = [
     "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",
@@ -166,4 +167,4 @@ omit = [
 ]
 
 [tool.coverage.report]
-show_missing = true
+show_missing = true

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

@@ -1892,7 +1892,8 @@ def create_instance(self) -> "PyCommandArguments":
                 id,
                 static_info.get("args"),
             )
-        except RuntimeError:
+        # Possible error thrown from the grpc layer
+        except (RuntimeError, ValueError):
             logger.warning(
                 "Create command arguments object is available from 23.1 onwards"
             )

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.protobuf.message import DecodeError, 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,25 @@ 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
+            try:
+                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):
+                                        new_ex_cls = cls
+                                        break
+            except DecodeError:
+                pass
+            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,13 @@ 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)
+
+
+@pytest.mark.fluent_version(">=25.2")
+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")