Merge pull request #1216 from robsdedude/feat/gql-error-cause-chain-helper

Introduce `GqlError.find_by_gql_status`

Author: robsdedude

docs/source/api.rst

@@ -1960,7 +1960,9 @@ GQL Errors
 ==========
 .. autoexception:: neo4j.exceptions.GqlError()
     :show-inheritance:
-    :members: gql_status, message, gql_status_description, gql_raw_classification, gql_classification, diagnostic_record, __cause__
+    :members:
+        gql_status, message, gql_status_description, gql_raw_classification, gql_classification, diagnostic_record,
+        find_by_gql_status, __cause__
 
 .. autoclass:: neo4j.exceptions.GqlErrorClassification()
     :show-inheritance:
@@ -2002,7 +2004,7 @@ Server-side errors
 
 .. autoexception:: neo4j.exceptions.Neo4jError()
     :show-inheritance:
-    :members: message, code, is_retriable, is_retryable
+    :members: message, code, is_retryable
 
 .. autoexception:: neo4j.exceptions.ClientError()
     :show-inheritance:

src/neo4j/exceptions.py

@@ -219,6 +219,14 @@ class GqlError(Exception):
     Instead, only subclasses are raised.
     Further, it is used as the :attr:`__cause__` of GqlError subclasses.
 
+    Sometimes it is helpful or necessary to traverse the cause chain of
+    GqlErrors to fully understand or appropriately handle the error. In such
+    cases, users can either traverse the :attr:`__cause__` attribute of the
+    error(s) or use the helper method :meth:`.find_by_gql_status`. Note that
+    :attr:`__cause__` is a standard attribute of all Python
+    :class:`BaseException` s: the cause chain may contain other exception types
+    besides GqlError.
+
     .. versionadded: 5.26
 
     .. versionchanged:: 6.0 Stabilized from preview.
@@ -233,6 +241,16 @@ class GqlError(Exception):
     _diagnostic_record: dict[str, _t.Any]  # copy to be used externally
     _gql_cause: GqlError | None
 
+    __cause__: BaseException | None
+    """
+    The GqlError's cause, if any.
+
+    Sometimes it is helpful or necessary to traverse the cause chain of
+    GqlErrors to fully understand or appropriately handle the error.
+
+    .. seealso:: :meth:`.find_by_gql_status`
+    """
+
     @staticmethod
     def _hydrate_cause(**metadata: _t.Any) -> GqlError:
         meta_extractor = _MetaExtractor(metadata)
@@ -427,6 +445,36 @@ def _get_status_diagnostic_record(self) -> dict[str, _t.Any]:
         self._status_diagnostic_record = dict(_UNKNOWN_GQL_DIAGNOSTIC_RECORD)
         return self._status_diagnostic_record
 
+    def find_by_gql_status(self, status: str) -> GqlError | None:
+        """
+        Return the first GqlError in the cause chain with the given GQL status.
+
+        This method traverses this GQLErorrs's :attr:`__cause__` chain,
+        starting with this error itself, and returns the first error that has
+        the given GQL status. If no error matches, :data:`None` is returned.
+
+        Example::
+
+            def invalid_syntax(err: GqlError) -> bool:
+                return err.find_by_gql_status("42001") is not None
+
+        :param status: The GQL status to search for.
+
+        :returns: The first matching error or :data:`None`.
+
+        .. versionadded:: 6.0
+        """
+        if self.gql_status == status:
+            return self
+
+        cause = self.__cause__
+        while cause is not None:
+            if isinstance(cause, GqlError) and cause.gql_status == status:
+                return cause
+            cause = getattr(cause, "__cause__", None)
+
+        return None
+
     def __str__(self):
         return (
             f"{{gql_status: {self.gql_status}}} "

tests/unit/common/test_exceptions.py

@@ -868,3 +868,35 @@ def test_deprecated_setter(attr):
         setattr(error, attr, obj)
 
     assert getattr(error, attr) is not obj
+
+
+@pytest.mark.parametrize("insert_after", range(-1, 3))
+def test_find_by_gql_status(insert_after: int) -> None:
+    root = error_to_find = None
+    if insert_after == -1:
+        root = error_to_find = _make_test_gql_error("12345")
+    for i in range(3):
+        root = _make_test_gql_error(f"{i + 2}2345", cause=root)
+        if i == insert_after:
+            root = error_to_find = _make_test_gql_error("12345", cause=root)
+
+    if root is None:
+        raise RuntimeError("unreachable, loop is not empty")
+    if error_to_find is None:
+        raise ValueError(
+            f"insert_after is out of range [-1, 3), got {insert_after}"
+        )
+
+    assert root.find_by_gql_status("12345") is error_to_find
+
+
+def test_find_by_gql_status_no_match() -> None:
+    root = None
+    for i in range(3):
+        root = _make_test_gql_error(f"{i + 1}2345", cause=root)
+
+    if root is None:
+        raise RuntimeError("unreachable, loop is not empty")
+
+    for status in ("2345", "02345", "42345", "54321"):
+        assert root.find_by_gql_status(status) is None