Cursor API

Author: technige

docs/source/index.rst

@@ -26,7 +26,7 @@ Session API
 
 .. autofunction:: neo4j.v1.record
 
-.. autoclass:: neo4j.v1.Result
+.. autoclass:: neo4j.v1.ResultCursor
    :members:
 
 .. autoclass:: neo4j.v1.ResultSummary

neo4j/v1/session.py

@@ -133,7 +133,7 @@ def recycle(self, session):
             pool.appendleft(session)
 
 
-class Result(list):
+class ResultCursor(object):
     """ A handler for the result of Cypher statement execution.
     """
 
@@ -143,60 +143,152 @@ class Result(list):
     #: Dictionary of parameters passed with the statement.
     parameters = None
 
-    def __init__(self, session, statement, parameters):
-        super(Result, self).__init__()
-        self.session = session
+    def __init__(self, connection, statement, parameters):
+        super(ResultCursor, self).__init__()
         self.statement = statement
         self.parameters = parameters
         self.keys = None
-        self.more = True
-        self.summary = None
-        self.bench_test = None
+        self._connection = connection
+        self._current = None
+        self._next = deque()
+        self._position = -1
+        self._summary = None
+        self._bench_test = None
+
+    def is_open(self):
+        """ Return ``True`` if this cursor is still open, ``False`` otherwise.
+        """
+        return bool(self._connection)
 
-    def on_header(self, metadata):
-        """ Called on receipt of the result header.
+    def close(self):
+        """ Consume the remainder of this result and detach the connection
+        from this cursor.
         """
-        self.keys = metadata["fields"]
-        if self.bench_test:
-            self.bench_test.start_recv = perf_counter()
+        if self._connection and not self._connection.closed:
+            self._consume()
+            self._connection = None
 
-    def on_record(self, values):
-        """ Called on receipt of each result record.
+    def next(self):
+        """ Advance to the next record, if available, and return a boolean
+        to indicate whether or not the cursor has moved.
         """
-        self.append(Record(self.keys, tuple(map(hydrated, values))))
+        if self._next:
+            values = self._next.popleft()
+            self._current = Record(self.keys, tuple(map(hydrated, values)))
+            self._position += 1
+            return True
+        elif self._summary:
+            return False
+        else:
+            self._connection.fetch_next()
+            return self.next()
 
-    def on_footer(self, metadata):
-        """ Called on receipt of the result footer.
+    def record(self):
+        """ Return the current record.
         """
-        self.more = False
-        self.summary = ResultSummary(self.statement, self.parameters, **metadata)
-        if self.bench_test:
-            self.bench_test.end_recv = perf_counter()
+        return self._current
 
-    def on_failure(self, metadata):
-        """ Called on execution failure.
+    def position(self):
+        """ Return the current cursor position.
         """
-        raise CypherError(metadata)
+        return self._position
 
-    def consume(self):
-        """ Consume the remainder of this result, triggering all appropriate
-        callback functions.
+    def at_end(self):
+        """ Return ``True`` if at the end of the record stream, ``False``
+        otherwise.
         """
-        fetch_next = self.session.connection.fetch_next
-        while self.more:
-            fetch_next()
+        if self._next:
+            return False
+        elif self._summary:
+            return True
+        else:
+            self._connection.fetch_next()
+            return self.at_end()
+
+    def records(self):
+        """ Yield all subsequent records.
+        """
+        while self.next():
+            yield self.record()
+
+    def __getitem__(self, item):
+        current = self._current
+        if current is None:
+            raise TypeError("No current record")
+        return current[item]
+
+    def get(self, item, default=None):
+        current = self._current
+        if current is None:
+            raise TypeError("No current record")
+        try:
+            return current[item]
+        except (IndexError, KeyError):
+            return default
 
     def summarize(self):
         """ Consume the remainder of this result and produce a summary.
 
         :rtype: ResultSummary
         """
-        self.consume()
-        return self.summary
+        self._consume()
+        return self._summary
+
+    def skip(self, records):
+        """ Move the cursor forward by a number of records.
+
+        :arg records: the number of records to step over
+        """
+        if records < 0:
+            raise ValueError("Cannot skip a negative number of records")
+        skipped = 0
+        while skipped < records and self.next():
+            skipped += 1
+        return skipped
+
+    def first(self):
+        """ Attempt to navigate to the first record in this stream, returning
+        ``True`` if this is possible, ``False`` otherwise.
+        """
+        if self._position < 0:
+            return self.next()
+        else:
+            return self._position == 0
+
+    def single(self):
+        """ Return ``True`` if able to navigate to first and only record.
+        """
+        return self.first() and self.at_end()
+
+    def _consume(self):
+        # Consume the remainder of this result, triggering all appropriate callback functions.
+        fetch_next = self._connection.fetch_next
+        while self._summary is None:
+            fetch_next()
+
+    def _on_header(self, metadata):
+        # Called on receipt of the result header.
+        self.keys = metadata["fields"]
+        if self._bench_test:
+            self._bench_test.start_recv = perf_counter()
+
+    def _on_record(self, values):
+        # Called on receipt of each result record.
+        self._next.append(values)
+
+    def _on_footer(self, metadata):
+        # Called on receipt of the result footer.
+        self._summary = ResultSummary(self.statement, self.parameters, **metadata)
+        if self._bench_test:
+            self._bench_test.end_recv = perf_counter()
+
+    def _on_failure(self, metadata):
+        # Called on execution failure.
+        raise CypherError(metadata)
 
 
 class ResultSummary(object):
-    """ A summary of execution returned with a :class:`.Result` object.
+    """ A summary of execution returned with a :class:`.ResultCursor` object.
     """
 
     #: The statement that was executed to produce this result.
@@ -391,7 +483,7 @@ def run(self, statement, parameters=None):
         :param statement: Cypher statement to execute
         :param parameters: dictionary of parameters
         :return: Cypher result
-        :rtype: :class:`.Result`
+        :rtype: :class:`.ResultCursor`
         """
 
         # Ensure the statement is a Unicode value
@@ -411,30 +503,28 @@ def run(self, statement, parameters=None):
         t = BenchTest()
         t.init = perf_counter()
 
-        result = Result(self, statement, parameters)
-        result.bench_test = t
+        cursor = ResultCursor(self.connection, statement, parameters)
+        cursor._bench_test = t
 
         run_response = Response(self.connection)
-        run_response.on_success = result.on_header
-        run_response.on_failure = result.on_failure
+        run_response.on_success = cursor._on_header
+        run_response.on_failure = cursor._on_failure
 
         pull_all_response = Response(self.connection)
-        pull_all_response.on_record = result.on_record
-        pull_all_response.on_success = result.on_footer
-        pull_all_response.on_failure = result.on_failure
+        pull_all_response.on_record = cursor._on_record
+        pull_all_response.on_success = cursor._on_footer
+        pull_all_response.on_failure = cursor._on_failure
 
         self.connection.append(RUN, (statement, parameters), response=run_response)
         self.connection.append(PULL_ALL, response=pull_all_response)
         t.start_send = perf_counter()
         self.connection.send()
         t.end_send = perf_counter()
 
-        result.consume()
-
         t.done = perf_counter()
         self.bench_tests.append(t)
 
-        return result
+        return cursor
 
     def close(self):
         """ If still usable, return this session to the driver pool it came from.

test/tck/tck_util.py

@@ -9,16 +9,16 @@
 
 def send_string(text):
     session = driver.session()
-    result = session.run(text)
+    cursor = session.run(text)
     session.close()
-    return result
+    return list(cursor.records())
 
 
 def send_parameters(statement, parameters):
     session = driver.session()
-    result = session.run(statement, parameters)
+    cursor = session.run(statement, parameters)
     session.close()
-    return result
+    return list(cursor.records())
 
 
 def get_bolt_value(type, value):

test/test_examples.py

@@ -37,7 +37,7 @@ def test_minimum_snippet(self):
         session = driver.session()
         session.run("CREATE (neo:Person {name:'Neo', age:23})")
 
-        for record in session.run("MATCH (p:Person) WHERE p.name = 'Neo' RETURN p.age"):
+        for record in session.run("MATCH (p:Person) WHERE p.name = 'Neo' RETURN p.age").records():
             print("Neo is {0} years old.".format(record["p.age"]))
         session.close()
         #end::minimum-snippet[]
@@ -46,8 +46,8 @@ def test_statement_with_parameters(self):
         driver = GraphDatabase.driver("bolt://localhost")
         session = driver.session()
         #tag::statement[]
-        result = session.run("CREATE (p:Person { name: {name} })", {"name": "The One"})
-        ones_created = result.summary.statistics.nodes_created
+        cursor = session.run("CREATE (p:Person { name: {name} })", {"name": "The One"})
+        ones_created = cursor.summarize().statistics.nodes_created
         print("There were {0} the ones created.".format(ones_created))
         #end::statement[]
         assert ones_created == 1
@@ -57,8 +57,8 @@ def test_statement_without_parameters(self):
         driver = GraphDatabase.driver("bolt://localhost")
         session = driver.session()
         #tag::statement-without-parameters[]
-        result = session.run("CREATE (p:Person { name: 'The One' })")
-        ones_created = result.summary.statistics.nodes_created
+        cursor = session.run("CREATE (p:Person { name: 'The One' })")
+        ones_created = cursor.summarize().statistics.nodes_created
         print("There were {0} the ones created.".format(ones_created))
         #end::statement-without-parameters[]
         assert ones_created == 1
@@ -72,8 +72,9 @@ def test_commit_a_transaction(self):
         tx.run("CREATE (p:Person { name: 'The One' })")
         tx.commit()
         #end::transaction-commit[]
-        res = session.run("MATCH (p:Person { name: 'The One' }) RETURN count(p)")
-        assert res[0]["count(p)"] == 1
+        cursor = session.run("MATCH (p:Person { name: 'The One' }) RETURN count(p)")
+        assert cursor.single()
+        assert cursor.record()["count(p)"] == 1
         session.close()
 
     def test_rollback_a_transaction(self):
@@ -84,8 +85,9 @@ def test_rollback_a_transaction(self):
         tx.run("CREATE (p:Person { name: 'The One' })")
         tx.rollback()
         #end::transaction-rollback[]
-        res = session.run("MATCH (p:Person { name: 'The One' }) RETURN count(p)")
-        assert res[0]["count(p)"] == 0
+        cursor = session.run("MATCH (p:Person { name: 'The One' }) RETURN count(p)")
+        assert cursor.single()
+        assert cursor.record()["count(p)"] == 0
         session.close()
 
     def test_require_encryption(self):