4.0 pull and discard (#388)

* pull n are used to fetch results in batches with the fetch_size configuration

* result.commit will now discard the rest if has more

* replaced the api result.summary() with result.consume()

* added new stub scripts

* fixed format issue

* using bolkit 4.0+ stub server

Before it was set to use the legacy bolkit stub server

* explicit version of boltkit in requirements

* fetches in batches

* added new stub server test scripts

* graceful shutdown for session result consume

* updated script to pass the bookmark to the next run

* added more time to let the stub servers start

* bolt 3 pull will now just ignore n and qid parameters

* added discard behaviour for transaction.consume and transaction.roll_back

Author: martin-neotech

docs/source/index.rst

@@ -157,6 +157,12 @@ Argument Renaming Changes
 * :code:`StatementResultSummary.protocol_version` is now :code:`ResultSummary.server.protocol_version`
 
 
+API Changes
+=========================
+
+* :code:`Result.summary()` have been replaced with :code:`Result.consume()`, this behaviour is to consume all remaining records in the buffer and returns the ResultSummary.
+
+
 Dependency Changes
 ==================
 

docs/source/results.rst

@@ -25,8 +25,6 @@ A :class:`.Result` is attached to an active connection, through a :class:`.Sessi
 
     .. automethod:: records
 
-    .. automethod:: summary
-
     .. automethod:: consume
 
     .. automethod:: single

neo4j/__init__.py

@@ -323,6 +323,7 @@ def close(self):
         """
         self._pool.close()
 
+    @experimental("The configuration may change in the future.")
     def verify_connectivity(self, **config):
         """ This verifies if the driver can connect to a remote server or a cluster
         by establishing a network connection with the remote and possibly exchanging
@@ -403,12 +404,14 @@ def pipeline(self, **config):
         PipelineConfig.consume(config)  # Consume the config
         return Pipeline(self._pool, pipeline_config)
 
+    @experimental("The configuration may change in the future.")
     def verify_connectivity(self, **config):
         server_agent = None
+        config["fetch_size"] = -1
         with self.session(**config) as session:
             result = session.run("RETURN 1 AS x")
             value = result.single().value()
-            summary = result.summary()
+            summary = result.consume()
             server_agent = summary.server.agent
         return server_agent
 
@@ -445,6 +448,7 @@ def pipeline(self, **config):
         PipelineConfig.consume(config)  # Consume the config
         return Pipeline(self._pool, pipeline_config)
 
+    @experimental("The configuration may change in the future.")
     def verify_connectivity(self, **config):
         """
         :raise ServiceUnavailable: raised if the server does not support routing or if routing support is broken.

neo4j/io/_bolt3.py

@@ -87,6 +87,7 @@ def __init__(self, unresolved_address, sock, max_connection_lifetime, *, auth=No
         self.responses = deque()
         self._max_connection_lifetime = max_connection_lifetime
         self._creation_timestamp = perf_counter()
+        self.state = None
 
         # Determine the user agent
         if user_agent:
@@ -173,18 +174,12 @@ def run(self, query, parameters=None, mode=None, bookmarks=None, metadata=None,
             self._append(b"\x10", fields, Response(self, **handlers))
 
     def discard(self, n=-1, qid=-1, **handlers):
-        if n != -1:
-            raise ValueError("Incremental discard is not supported in Bolt 3")
-        if qid != -1:
-            raise ValueError("Query selection on discard is not supported in Bolt 3")
+        # Just ignore n and qid, it is not supported in the Bolt 3 Protocol.
         log.debug("[#%04X]  C: DISCARD_ALL", self.local_port)
         self._append(b"\x2F", (), Response(self, **handlers))
 
     def pull(self, n=-1, qid=-1, **handlers):
-        if n != -1:
-            raise ValueError("Incremental pull is not supported in Bolt 3")
-        if qid != -1:
-            raise ValueError("Query selection on pull is not supported in Bolt 3")
+        # Just ignore n and qid, it is not supported in the Bolt 3 Protocol.
         log.debug("[#%04X]  C: PULL_ALL", self.local_port)
         self._append(b"\x3F", (), Response(self, **handlers))
 
@@ -305,7 +300,7 @@ def fetch_message(self):
             raise
 
         if details:
-            log.debug("[#%04X]  S: RECORD * %d", self.local_port, len(details))  # TODO
+            log.debug("[#%04X]  S: RECORD * %d", self.local_port, len(details))
             self.responses[0].on_records(details)
 
         if summary_signature is None:

neo4j/io/_bolt4x0.py

@@ -86,6 +86,7 @@ def __init__(self, unresolved_address, sock, max_connection_lifetime, *, auth=No
         self.responses = deque()
         self._max_connection_lifetime = max_connection_lifetime  # self.pool_config.max_connection_lifetime
         self._creation_timestamp = perf_counter()
+        self.state = None
 
         # Determine the user agent
         if user_agent:
@@ -303,7 +304,7 @@ def fetch_message(self):
             raise
 
         if details:
-            log.debug("[#%04X]  S: RECORD * %d", self.local_port, len(details))  # TODO
+            log.debug("[#%04X]  S: RECORD * %d", self.local_port, len(details))
             self.responses[0].on_records(details)
 
         if summary_signature is None:
@@ -482,19 +483,30 @@ def __init__(self, connection, **handlers):
     def on_records(self, records):
         """ Called when one or more RECORD messages have been received.
         """
+        self.connection.state = "streaming"
         handler = self.handlers.get("on_records")
         if callable(handler):
             handler(records)
 
     def on_success(self, metadata):
         """ Called when a SUCCESS message has been received.
         """
-        handler = self.handlers.get("on_success")
-        if callable(handler):
-            handler(metadata)
-        handler = self.handlers.get("on_summary")
-        if callable(handler):
-            handler()
+        if metadata.get("has_more"):
+            if self.connection.state == "streaming_discard_all":
+                handler = self.handlers.get("on_success_has_more_streaming_discard_all")
+                self.connection.state = None
+                if callable(handler):
+                    handler(self.connection, **self.handlers)
+            else:
+                self.connection.state = "streaming_has_more"
+        else:
+            self.connection.state = None
+            handler = self.handlers.get("on_success")
+            if callable(handler):
+                handler(metadata)
+            handler = self.handlers.get("on_summary")
+            if callable(handler):
+                handler()
 
     def on_failure(self, metadata):
         """ Called when a FAILURE message has been received.