Raise hydration errors on value access

Author: robsdedude

docs/source/api.rst

@@ -1248,7 +1248,6 @@ Neo4j Execution Errors
 .. autoclass:: neo4j.exceptions.Neo4jError
     :members: message, code, is_retriable, is_retryable
 
-
 .. autoclass:: neo4j.exceptions.ClientError
     :show-inheritance:
 
@@ -1330,6 +1329,9 @@ Connectivity Errors
 .. autoclass:: neo4j.exceptions.ReadServiceUnavailable
     :show-inheritance:
 
+.. autoclass:: neo4j.exceptions.BrokenRecordError
+    :show-inheritance:
+
 .. autoclass:: neo4j.exceptions.ConfigurationError
     :show-inheritance:
 

neo4j/__init__.py

@@ -108,7 +108,6 @@
     "BoltDriver",
     "Bookmark",
     "Bookmarks",
-    "Config",
     "custom_auth",
     "DEFAULT_DATABASE",
     "Driver",
@@ -120,15 +119,13 @@
     "kerberos_auth",
     "ManagedTransaction",
     "Neo4jDriver",
-    "PoolConfig",
     "Query",
     "READ_ACCESS",
     "Record",
     "Result",
     "ResultSummary",
     "ServerInfo",
     "Session",
-    "SessionConfig",
     "SummaryCounters",
     "Transaction",
     "TRUST_ALL_CERTIFICATES",
@@ -138,7 +135,6 @@
     "TrustSystemCAs",
     "unit_of_work",
     "Version",
-    "WorkspaceConfig",
     "WRITE_ACCESS",
 ]
 

neo4j/_async/work/result.py

@@ -20,6 +20,7 @@
 from warnings import warn
 
 from ..._async_compat.util import AsyncUtil
+from ..._codec.hydration import BrokenHydrationObject
 from ..._data import (
     Record,
     RecordTableRowExporter,
@@ -145,6 +146,11 @@ async def on_failed_attach(metadata):
     def _pull(self):
         def on_records(records):
             if not self._discarding:
+                records = (
+                    record.raw_data
+                    if isinstance(record, BrokenHydrationObject) else record
+                    for record in records
+                )
                 self._record_buffer.extend((
                     Record(zip(self._keys, record))
                     for record in records

neo4j/_codec/hydration/__init__.py

@@ -15,11 +15,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ._common import HydrationScope
+from ._common import (
+    BrokenHydrationObject,
+    HydrationScope,
+)
 from ._interface import HydrationHandlerABC
 
 
 __all__ = [
+    "BrokenHydrationObject",
     "HydrationHandlerABC",
     "HydrationScope",
 ]

neo4j/_codec/hydration/_common.py

@@ -16,18 +16,46 @@
 # limitations under the License.
 
 
+from copy import copy
+
 from ...graph import Graph
 from ..packstream import Structure
 
 
+class BrokenHydrationObject:
+    """
+    Represents an object from the server, not understood by the driver.
+
+    A :class:`neo4j.Record` might contain a ``BrokenHydrationObject`` object
+    if the driver received data from the server that it did not understand.
+    This can for instance happen if the server sends a zoned datetime with a
+    zone name unknown to the driver.
+
+    There is no need to explicitly check for this type. Any method on the
+    :class:`neo4j.Record` that would return a ``BrokenHydrationObject``, will
+    raise a :exc:`neo4j.exceptions.BrokenRecordError`
+    with the original exception as cause.
+    """
+
+    def __init__(self, error, raw_data):
+        self.error = error
+        "The exception raised while decoding the received object."
+        self.raw_data = raw_data
+        """The raw data that caused the exception."""
+
+    def exception_copy(self):
+        exc_copy = copy(self.error)
+        exc_copy.with_traceback(self.error.__traceback__)
+        return exc_copy
+
+
 class GraphHydrator:
     def __init__(self):
         self.graph = Graph()
         self.struct_hydration_functions = {}
 
 
 class HydrationScope:
-
     def __init__(self, hydration_handler, graph_hydrator):
         self._hydration_handler = hydration_handler
         self._graph_hydrator = graph_hydrator
@@ -37,14 +65,35 @@ def __init__(self, hydration_handler, graph_hydrator):
         }
         self.hydration_hooks = {
             Structure: self._hydrate_structure,
+            list: self._hydrate_list,
+            dict: self._hydrate_dict,
         }
         self.dehydration_hooks = hydration_handler.dehydration_functions
 
     def _hydrate_structure(self, value):
         f = self._struct_hydration_functions.get(value.tag)
-        if not f:
-            return value
-        return f(*value.fields)
+        try:
+            if not f:
+                raise ValueError(
+                    f"Protocol error: unknown Structure tag: {value.tag!r}"
+                )
+            return f(*value.fields)
+        except Exception as e:
+            return BrokenHydrationObject(e, value)
+
+    @staticmethod
+    def _hydrate_list(value):
+        for v in value:
+            if isinstance(v, BrokenHydrationObject):
+                return BrokenHydrationObject(v.error, value)
+        return value
+
+    @staticmethod
+    def _hydrate_dict(value):
+        for v in value.values():
+            if isinstance(v, BrokenHydrationObject):
+                return BrokenHydrationObject(v.error, value)
+        return value
 
     def get_graph(self):
         return self._graph_hydrator.graph

neo4j/_data.py

@@ -28,7 +28,10 @@
 from functools import reduce
 from operator import xor as xor_operator
 
+from ._codec.hydration import BrokenHydrationObject
 from ._conf import iter_items
+from ._meta import deprecated
+from .exceptions import BrokenRecordError
 from .graph import (
     Node,
     Path,
@@ -55,9 +58,26 @@ def __new__(cls, iterable=()):
         inst.__keys = tuple(keys)
         return inst
 
+    def _broken_record_error(self, index):
+        return BrokenRecordError(
+            f"Record contains broken data at {index} ('{self.__keys[index]}')"
+        )
+
+    def _super_getitem_single(self, index):
+        value = super().__getitem__(index)
+        if isinstance(value, BrokenHydrationObject):
+            raise self._broken_record_error(index) from value.error
+        return value
+
     def __repr__(self):
-        return "<%s %s>" % (self.__class__.__name__,
-                            " ".join("%s=%r" % (field, self[i]) for i, field in enumerate(self.__keys)))
+        return "<%s %s>" % (
+            self.__class__.__name__,
+            " ".join("%s=%r" % (field, value)
+                     for field, value in zip(self.__keys, super().__iter__()))
+        )
+
+    def __str__(self):
+        return self.__repr__()
 
     def __eq__(self, other):
         """ In order to be flexible regarding comparison, the equality rules
@@ -83,18 +103,26 @@ def __ne__(self, other):
     def __hash__(self):
         return reduce(xor_operator, map(hash, self.items()))
 
+    def __iter__(self):
+        for i, v in enumerate(super().__iter__()):
+            if isinstance(v, BrokenHydrationObject):
+                raise self._broken_record_error(i) from v.error
+            yield v
+
     def __getitem__(self, key):
         if isinstance(key, slice):
             keys = self.__keys[key]
-            values = super(Record, self).__getitem__(key)
+            values = super().__getitem__(key)
             return self.__class__(zip(keys, values))
         try:
             index = self.index(key)
         except IndexError:
             return None
         else:
-            return super(Record, self).__getitem__(index)
+            return self._super_getitem_single(index)
 
+    # TODO: 6.0 - remove
+    @deprecated("This method is deprecated and will be removed in the future.")
     def __getslice__(self, start, stop):
         key = slice(start, stop)
         keys = self.__keys[key]
@@ -114,7 +142,7 @@ def get(self, key, default=None):
         except ValueError:
             return default
         if 0 <= index < len(self):
-            return super(Record, self).__getitem__(index)
+            return self._super_getitem_single(index)
         else:
             return default
 
@@ -197,7 +225,8 @@ def items(self, *keys):
                 else:
                     d.append((self.__keys[i], self[i]))
             return d
-        return list((self.__keys[i], super(Record, self).__getitem__(i)) for i in range(len(self)))
+        return list((self.__keys[i], self._super_getitem_single(i))
+                    for i in range(len(self)))
 
     def data(self, *keys):
         """ Return the keys and values of this record as a dictionary,

neo4j/_sync/work/result.py

@@ -20,6 +20,7 @@
 from warnings import warn
 
 from ..._async_compat.util import Util
+from ..._codec.hydration import BrokenHydrationObject
 from ..._data import (
     Record,
     RecordTableRowExporter,
@@ -145,6 +146,11 @@ def on_failed_attach(metadata):
     def _pull(self):
         def on_records(records):
             if not self._discarding:
+                records = (
+                    record.raw_data
+                    if isinstance(record, BrokenHydrationObject) else record
+                    for record in records
+                )
                 self._record_buffer.extend((
                     Record(zip(self._keys, record))
                     for record in records

neo4j/exceptions.py

@@ -396,6 +396,14 @@ def is_retryable(self):
         return True
 
 
+class BrokenRecordError(DriverError):
+    """ Raised when accessing a Record's field that couldn't be decoded.
+
+    This can for instance happen when the server sends a zoned datetime with a
+    zone id unknown to the client.
+    """
+
+
 class RoutingServiceUnavailable(ServiceUnavailable):
     """ Raised when no routing service is available.
     """

tests/unit/async_/work/test_result.py

@@ -55,18 +55,18 @@ def __init__(self, fields, records):
         self.fields = tuple(fields)
         self.hydration_scope = HydrationHandler().new_hydration_scope()
         self.records = tuple(records)
-        self._hydrate_records()
-
         assert all(len(self.fields) == len(r) for r in self.records)
 
+        self._hydrate_records()
+
     def _hydrate_records(self):
         def _hydrate(value):
+            if isinstance(value, (list, tuple)):
+                value = type(value)(_hydrate(v) for v in value)
+            elif isinstance(value, dict):
+                value = {k: _hydrate(v) for k, v in value.items()}
             if type(value) in self.hydration_scope.hydration_hooks:
                 return self.hydration_scope.hydration_hooks[type(value)](value)
-            if isinstance(value, (list, tuple)):
-                return type(value)(_hydrate(v) for v in value)
-            if isinstance(value, dict):
-                return {k: _hydrate(v) for k, v in value.items()}
             return value
 
         self.records = tuple(_hydrate(r) for r in self.records)
@@ -605,7 +605,6 @@ async def test_data(num_records):
         assert record.data.called_once_with("hello", "world")
 
 
-# TODO: dehydration now happens on a much lower level
 @pytest.mark.parametrize("records", (
     Records(["n"], []),
     Records(["n"], [[42], [69], [420], [1337]]),
@@ -621,19 +620,9 @@ async def test_data(num_records):
     ]),
 ))
 @mark_async_test
-async def test_result_graph(records, async_scripted_connection):
-    async_scripted_connection.set_script((
-        ("run", {"on_success": ({"fields": records.fields},),
-                 "on_summary": None}),
-        ("pull", {
-            "on_records": (records.records,),
-            "on_success": None,
-            "on_summary": None
-        }),
-    ))
-    async_scripted_connection.new_hydration_scope.return_value = \
-        records.hydration_scope
-    result = AsyncResult(async_scripted_connection, 1, noop, noop)
+async def test_result_graph(records):
+    connection = AsyncConnectionStub(records=records)
+    result = AsyncResult(connection, 1, noop, noop)
     await result._run("CYPHER", {}, None, None, "r", None)
     graph = await result.graph()
     assert isinstance(graph, Graph)
@@ -1095,3 +1084,22 @@ async def test_to_df_parse_dates(keys, values, expected_df, expand):
         df = await result.to_df(expand=expand, parse_dates=True)
 
     pd.testing.assert_frame_equal(df, expected_df)
+
+
+@pytest.mark.parametrize("nested", [True, False])
+@mark_async_test
+async def test_broken_hydration(nested):
+    value_in = Structure(b"a", "broken")
+    if nested:
+        value_in = [value_in]
+    records_in = Records(["foo", "bar"], [["foobar", value_in]])
+    connection = AsyncConnectionStub(records=records_in)
+    result = AsyncResult(connection, 1, noop, noop)
+    await result._run("CYPHER", {}, None, None, "r", None)
+    records_out = await AsyncUtil.list(result)
+    assert len(records_out) == 1
+    record_out = records_out[0]
+    assert len(record_out) == 2
+    assert record_out[0] == "foobar"
+    with pytest.raises(ValueError):
+        record_out[1]