Split session tx result prototype (#390)

* Non-working at all...

* Babysteps on result before weekend..

* Result prototype

* Multiple results in tx

* Pair programming

* Removed commented code

* integration test is runable

* Bookmarks prototype

* Small fixes

* Wrong func name

* Syntax fixes

* Syntax error

* correct integration tests for multiple results

* explicit transaction case1 is runable

* message fetching logic needs to be fixed

* Better conn handling

* No fetch in tx _begin

* nested results are working now for explicit transactions.

* Fixed some tests

* transaction.commit will properly consume all the records

* tests/integration/test_autocommit.py is green

* Fixed bookmarks api

* fixed a case of fetching records where there are no records to fetch.

* integration test_result is green

* Bookmark tests green

* data is only used on record objects

* Removed 2s sleep for stub tests

* fixed an integration test

test_errors_on_run_transaction

* Rollback on failed connection fix

* added is_reset state to bolt4

* stabilized an integration test

* added check for Query object for transaction.run

* Handle broken connections

* fixed case with Bolt 3 tx.run tx.run case

* cleanup of code

* logging cleanup

* comment removed cleanup

* removal of old comment

* fixed raise hell error

* fixed case with session.run session.run

* fixed session.run session.run

* fixed Transaction class api

* fixed the api for Result object

* fixed api for Session Object

* removed _detach

replaced calls with consume and _buffer_all for correct behaviour

Co-authored-by: Peter Wilhelmsson <peter.wilhelmsson@neo4j.com>

Author: martin-neotech

neo4j/__init__.py

@@ -93,13 +93,19 @@
     Record,
 )
 from neo4j.work.simple import (
-    Transaction,
-    Result,
-    ResultSummary,
     Query,
     Session,
     unit_of_work,
 )
+from neo4j.work.transaction import (
+    Transaction,
+)
+from neo4j.work.result import (
+    Result,
+)
+from neo4j.work.summary import (
+    ResultSummary,
+)
 
 
 log = getLogger("neo4j")

neo4j/io/_bolt3.py

@@ -87,7 +87,8 @@ 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
+        self.supports_multiple_results = False
+        self._is_reset = True
 
         # Determine the user agent
         if user_agent:
@@ -172,6 +173,7 @@ def run(self, query, parameters=None, mode=None, bookmarks=None, metadata=None,
             self._append(b"\x10", fields, CommitResponse(self, **handlers))
         else:
             self._append(b"\x10", fields, Response(self, **handlers))
+        self._is_reset = False
 
     def discard(self, n=-1, qid=-1, **handlers):
         # Just ignore n and qid, it is not supported in the Bolt 3 Protocol.
@@ -182,6 +184,7 @@ def pull(self, n=-1, qid=-1, **handlers):
         # 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))
+        self._is_reset = False
 
     def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None, db=None, **handlers):
         if db is not None:
@@ -206,6 +209,7 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None, db=None,
                 raise TypeError("Timeout must be specified as a number of seconds")
         log.debug("[#%04X]  C: BEGIN %r", self.local_port, extra)
         self._append(b"\x11", (extra,), Response(self, **handlers))
+        self._is_reset = False
 
     def commit(self, **handlers):
         log.debug("[#%04X]  C: COMMIT", self.local_port)
@@ -239,6 +243,7 @@ def fail(metadata):
         self._append(b"\x0F", response=Response(self, on_failure=fail))
         self.send_all()
         self.fetch_all()
+        self._is_reset = True
 
     def _send_all(self):
         data = self.outbox.view()
@@ -300,7 +305,7 @@ def fetch_message(self):
             raise
 
         if details:
-            log.debug("[#%04X]  S: RECORD * %d", self.local_port, len(details))
+            log.debug("[#%04X]  S: RECORD * %d", self.local_port, len(details))  # Do not log any data
             self.responses[0].on_records(details)
 
         if summary_signature is None:
@@ -529,4 +534,4 @@ def on_failure(self, metadata):
 
 class CommitResponse(Response):
 
-    pass
+    pass

neo4j/io/_bolt4x0.py

@@ -86,7 +86,8 @@ 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
+        self.supports_multiple_results = True
+        self._is_reset = True
 
         # Determine the user agent
         if user_agent:
@@ -171,6 +172,7 @@ def run(self, query, parameters=None, mode=None, bookmarks=None, metadata=None,
             self._append(b"\x10", fields, CommitResponse(self, **handlers))
         else:
             self._append(b"\x10", fields, Response(self, **handlers))
+        self._is_reset = False
 
     def discard(self, n=-1, qid=-1, **handlers):
         extra = {"n": n}
@@ -185,6 +187,7 @@ def pull(self, n=-1, qid=-1, **handlers):
             extra["qid"] = qid
         log.debug("[#%04X]  C: PULL %r", self.local_port, extra)
         self._append(b"\x3F", (extra,), Response(self, **handlers))
+        self._is_reset = False
 
     def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
               db=None, **handlers):
@@ -210,6 +213,7 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
                 raise TypeError("Timeout must be specified as a number of seconds")
         log.debug("[#%04X]  C: BEGIN %r", self.local_port, extra)
         self._append(b"\x11", (extra,), Response(self, **handlers))
+        self._is_reset = False
 
     def commit(self, **handlers):
         log.debug("[#%04X]  C: COMMIT", self.local_port)
@@ -243,6 +247,7 @@ def fail(metadata):
         self._append(b"\x0F", response=Response(self, on_failure=fail))
         self.send_all()
         self.fetch_all()
+        self._is_reset = True
 
     def _send_all(self):
         data = self.outbox.view()
@@ -304,7 +309,7 @@ def fetch_message(self):
             raise
 
         if details:
-            log.debug("[#%04X]  S: RECORD * %d", self.local_port, len(details))
+            log.debug("[#%04X]  S: RECORD * %d", self.local_port, len(details))  # Do not log any data
             self.responses[0].on_records(details)
 
         if summary_signature is None:
@@ -483,27 +488,18 @@ 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.
         """
-        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_success")
+        if callable(handler):
+            handler(metadata)
+
+        if not metadata.get("has_more"):
             handler = self.handlers.get("on_summary")
             if callable(handler):
                 handler()
@@ -544,4 +540,4 @@ def on_failure(self, metadata):
 
 class CommitResponse(Response):
 
-    pass
+    pass