Check Result scope and raise RuntimeError if appropriate

Results are tied to transactions (except auto-commit transactions).
When the transaction ends, the result becomes useless. Raising instead of
silently ignoring this fact will help developers to find potential bugs faster.

After consuming a Result, there can never be records left. There is meaning in
trying to obtain them. Hence, we raise a RuntimeError to make the user aware of
potentially wrong code.

Author: robsdedude

CHANGELOG.md

@@ -5,7 +5,9 @@
 - Python 3.10 support added
 - Python 3.6 support has been dropped.
 - `Result`, `Session`, and `Transaction` can no longer be imported from
-  `neo4j.work`. They should've been imported from `neo4j` all along.
+  `neo4j.work`. They should've been imported from `neo4j` all along.  
+  Remark: It's recommended to import everything needed directly from `noe4j`,
+  not its submodules or subpackages.
 - Experimental pipelines feature has been removed.
 - Experimental async driver has been added.
 - `ResultSummary.server.version_info` has been removed.  
@@ -54,6 +56,17 @@
   was silently ignored.
 - `Result.single` now raises `ResultNotSingleError` if not exactly one result is
   available.
+- Result scope:  
+  - Records of Results cannot be accessed (`peek`, `single`, `iter`, ...)
+    after their owning transaction has been closed, committed, or rolled back.
+    Previously, this would yield undefined behavior.
+    It now raises a `RuntimeError`.
+  - Records of Results cannot be accessed (`peek`, `single`, `iter`, ...)
+    after the Result has been consumed (`Result.consume()`).
+    Previously, this would always yield no records.
+    It now raises a `RuntimeError`.
+  - New method `Result.closed()` can be used to check for this condition if
+    necessary.
 
 ## Version 4.4
 

docs/source/api.rst

@@ -765,6 +765,8 @@ A :class:`neo4j.Result` is attached to an active connection, through a :class:`n
 
     .. automethod:: data
 
+    .. automethod:: closed
+
 See https://neo4j.com/docs/driver-manual/current/cypher-workflow/#driver-type-mapping for more about type mapping.
 
 

docs/source/async_api.rst

@@ -499,4 +499,6 @@ A :class:`neo4j.AsyncResult` is attached to an active connection, through a :cla
 
     .. automethod:: data
 
+    .. automethod:: closed
+
 See https://neo4j.com/docs/driver-manual/current/cypher-workflow/#driver-type-mapping for more about type mapping.

neo4j/_async/work/result.py

@@ -26,6 +26,17 @@
 from ..io import ConnectionErrorHandler
 
 
+_RESULT_OUT_OF_SCOPE_ERROR = (
+    "The result is out of scope. The associated transaction "
+    "has been closed. Results can only be used while the "
+    "transaction is open."
+)
+_RESULT_CONSUMED_ERROR = (
+    "The result has been consumed. Fetch all needed records before calling "
+    "Result.consume()."
+)
+
+
 class AsyncResult:
     """A handler for the result of Cypher query execution. Instances
     of this class are typically constructed and returned by
@@ -54,6 +65,10 @@ def __init__(self, connection, hydrant, fetch_size, on_closed,
         self._has_more = False
         # the result has been fully iterated or consumed
         self._closed = False
+        # the result has been consumed
+        self._consumed = False
+        # the result has been closed as a result of closing the transaction
+        self._out_of_scope = False
 
     @property
     def _qid(self):
@@ -196,6 +211,10 @@ async def __aiter__(self):
                 await self._connection.send_all()
 
         self._closed = True
+        if self._out_of_scope:
+            raise RuntimeError(_RESULT_OUT_OF_SCOPE_ERROR)
+        if self._consumed:
+            raise RuntimeError(_RESULT_CONSUMED_ERROR)
 
     async def __anext__(self):
         return await self.__aiter__().__anext__()
@@ -216,6 +235,10 @@ async def _buffer(self, n=None):
         Might ent up with fewer records in the buffer if there are not enough
         records available.
         """
+        if self._out_of_scope:
+            raise RuntimeError(_RESULT_OUT_OF_SCOPE_ERROR)
+        if self._consumed:
+            raise RuntimeError(_RESULT_CONSUMED_ERROR)
         if n is not None and len(self._record_buffer) >= n:
             return
         record_buffer = deque()
@@ -261,6 +284,14 @@ def keys(self):
         """
         return self._keys
 
+    async def _tx_end(self):
+        # Handle closure of the associated transaction.
+        #
+        # This will consume the result and mark it at out of scope.
+        # Subsequent calls to `next` will raise a RuntimeError.
+        await self.consume()
+        self._out_of_scope = True
+
     async def consume(self):
         """Consume the remainder of this result and return a :class:`neo4j.ResultSummary`.
 
@@ -302,7 +333,9 @@ async def get_two_tx(tx):
             async for _ in self:
                 pass
 
-        return self._obtain_summary()
+        summary = self._obtain_summary()
+        self._consumed = True
+        return summary
 
     async def single(self):
         """Obtain the next and only remaining record from this result if available else return None.
@@ -312,7 +345,10 @@ async def single(self):
         the first of these is still returned.
 
         :returns: the next :class:`neo4j.AsyncRecord`.
-        :raises: ResultNotSingleError if not exactly one record is available.
+
+        :raises ResultNotSingleError: if not exactly one record is available.
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed.
         """
         await self._buffer(2)
         if not self._record_buffer:
@@ -332,6 +368,10 @@ async def peek(self):
         This leaves the record in the buffer for further processing.
 
         :returns: the next :class:`.Record` or :const:`None` if none remain
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed or the Result has been explicitly
+            consumed.
         """
         await self._buffer(1)
         if self._record_buffer:
@@ -344,6 +384,10 @@ async def graph(self):
 
         :returns: a result graph
         :rtype: :class:`neo4j.graph.Graph`
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed or the Result has been explicitly
+            consumed.
         """
         await self._buffer_all()
         return self._hydrant.graph
@@ -355,8 +399,13 @@ async def value(self, key=0, default=None):
 
         :param key: field to return for each remaining record. Obtain a single value from the record by index or key.
         :param default: default value, used if the index of key is unavailable
+
         :returns: list of individual values
         :rtype: list
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed or the Result has been explicitly
+            consumed.
         """
         return [record.value(key, default) async for record in self]
 
@@ -366,8 +415,13 @@ async def values(self, *keys):
         See :class:`neo4j.AsyncRecord.values`
 
         :param keys: fields to return for each remaining record. Optionally filtering to include only certain values by index or key.
+
         :returns: list of values lists
         :rtype: list
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed or the Result has been explicitly
+            consumed.
         """
         return [record.values(*keys) async for record in self]
 
@@ -377,7 +431,26 @@ async def data(self, *keys):
         See :class:`neo4j.AsyncRecord.data`
 
         :param keys: fields to return for each remaining record. Optionally filtering to include only certain values by index or key.
+
         :returns: list of dictionaries
         :rtype: list
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed.
         """
         return [record.data(*keys) async for record in self]
+
+    def closed(self):
+        """Return True if the result is still valid (not closed).
+
+        When a result gets consumed :meth:`consume` or the transaction that
+        owns the result gets closed (committed, rolled back, closed), the
+        result cannot be used to acquire further records.
+
+        In such case, all methods that need to access the Result's records,
+        will raise a :exc:`RuntimeError` when called.
+
+        :returns: whether the result is closed.
+        :rtype: bool
+        """
+        return self._out_of_scope or self._consumed

neo4j/_async/work/transaction.py

@@ -78,7 +78,7 @@ async def _error_handler(self, exc):
 
     async def _consume_results(self):
         for result in self._results:
-            await result.consume()
+            await result._tx_end()
         self._results = []
 
     async def run(self, query, parameters=None, **kwparameters):

neo4j/_sync/work/result.py

@@ -26,6 +26,17 @@
 from ..io import ConnectionErrorHandler
 
 
+_RESULT_OUT_OF_SCOPE_ERROR = (
+    "The result is out of scope. The associated transaction "
+    "has been closed. Results can only be used while the "
+    "transaction is open."
+)
+_RESULT_CONSUMED_ERROR = (
+    "The result has been consumed. Fetch all needed records before calling "
+    "Result.consume()."
+)
+
+
 class Result:
     """A handler for the result of Cypher query execution. Instances
     of this class are typically constructed and returned by
@@ -54,6 +65,10 @@ def __init__(self, connection, hydrant, fetch_size, on_closed,
         self._has_more = False
         # the result has been fully iterated or consumed
         self._closed = False
+        # the result has been consumed
+        self._consumed = False
+        # the result has been closed as a result of closing the transaction
+        self._out_of_scope = False
 
     @property
     def _qid(self):
@@ -196,6 +211,10 @@ def __iter__(self):
                 self._connection.send_all()
 
         self._closed = True
+        if self._out_of_scope:
+            raise RuntimeError(_RESULT_OUT_OF_SCOPE_ERROR)
+        if self._consumed:
+            raise RuntimeError(_RESULT_CONSUMED_ERROR)
 
     def __next__(self):
         return self.__iter__().__next__()
@@ -216,6 +235,10 @@ def _buffer(self, n=None):
         Might ent up with fewer records in the buffer if there are not enough
         records available.
         """
+        if self._out_of_scope:
+            raise RuntimeError(_RESULT_OUT_OF_SCOPE_ERROR)
+        if self._consumed:
+            raise RuntimeError(_RESULT_CONSUMED_ERROR)
         if n is not None and len(self._record_buffer) >= n:
             return
         record_buffer = deque()
@@ -261,6 +284,14 @@ def keys(self):
         """
         return self._keys
 
+    def _tx_end(self):
+        # Handle closure of the associated transaction.
+        #
+        # This will consume the result and mark it at out of scope.
+        # Subsequent calls to `next` will raise a RuntimeError.
+        self.consume()
+        self._out_of_scope = True
+
     def consume(self):
         """Consume the remainder of this result and return a :class:`neo4j.ResultSummary`.
 
@@ -302,7 +333,9 @@ def get_two_tx(tx):
             for _ in self:
                 pass
 
-        return self._obtain_summary()
+        summary = self._obtain_summary()
+        self._consumed = True
+        return summary
 
     def single(self):
         """Obtain the next and only remaining record from this result if available else return None.
@@ -312,7 +345,10 @@ def single(self):
         the first of these is still returned.
 
         :returns: the next :class:`neo4j.Record`.
-        :raises: ResultNotSingleError if not exactly one record is available.
+
+        :raises ResultNotSingleError: if not exactly one record is available.
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed.
         """
         self._buffer(2)
         if not self._record_buffer:
@@ -332,6 +368,10 @@ def peek(self):
         This leaves the record in the buffer for further processing.
 
         :returns: the next :class:`.Record` or :const:`None` if none remain
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed or the Result has been explicitly
+            consumed.
         """
         self._buffer(1)
         if self._record_buffer:
@@ -344,6 +384,10 @@ def graph(self):
 
         :returns: a result graph
         :rtype: :class:`neo4j.graph.Graph`
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed or the Result has been explicitly
+            consumed.
         """
         self._buffer_all()
         return self._hydrant.graph
@@ -355,8 +399,13 @@ def value(self, key=0, default=None):
 
         :param key: field to return for each remaining record. Obtain a single value from the record by index or key.
         :param default: default value, used if the index of key is unavailable
+
         :returns: list of individual values
         :rtype: list
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed or the Result has been explicitly
+            consumed.
         """
         return [record.value(key, default) for record in self]
 
@@ -366,8 +415,13 @@ def values(self, *keys):
         See :class:`neo4j.Record.values`
 
         :param keys: fields to return for each remaining record. Optionally filtering to include only certain values by index or key.
+
         :returns: list of values lists
         :rtype: list
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed or the Result has been explicitly
+            consumed.
         """
         return [record.values(*keys) for record in self]
 
@@ -377,7 +431,26 @@ def data(self, *keys):
         See :class:`neo4j.Record.data`
 
         :param keys: fields to return for each remaining record. Optionally filtering to include only certain values by index or key.
+
         :returns: list of dictionaries
         :rtype: list
+
+        :raises RuntimeError: if the transaction from which this result was
+            obtained has been closed.
         """
         return [record.data(*keys) for record in self]
+
+    def closed(self):
+        """Return True if the result is still valid (not closed).
+
+        When a result gets consumed :meth:`consume` or the transaction that
+        owns the result gets closed (committed, rolled back, closed), the
+        result cannot be used to acquire further records.
+
+        In such case, all methods that need to access the Result's records,
+        will raise a :exc:`RuntimeError` when called.
+
+        :returns: whether the result is closed.
+        :rtype: bool
+        """
+        return self._out_of_scope or self._consumed

neo4j/_sync/work/transaction.py

@@ -78,7 +78,7 @@ def _error_handler(self, exc):
 
     def _consume_results(self):
         for result in self._results:
-            result.consume()
+            result._tx_end()
         self._results = []
 
     def run(self, query, parameters=None, **kwparameters):