Timeout and metadata for explicit transactions

Author: technige

neo4j/__init__.py

@@ -450,12 +450,20 @@ def fail(_):
             "server": server,
             "protocol_version": protocol_version,
         }
+
+        def done(summary_metadata):
+            metadata.update(summary_metadata)
+            bookmark = metadata.get("bookmark")
+            if bookmark:
+                self._bookmarks_in = tuple([bookmark])
+                self._bookmark_out = bookmark
+
         self._last_result = result = BoltStatementResult(self, hydrant, metadata)
         cx.run(statement, parameters, on_success=metadata.update, on_failure=fail)
         cx.pull_all(
             on_records=lambda records: result._records.extend(
                 hydrant.hydrate_records(result.keys(), records)),
-            on_success=metadata.update,
+            on_success=done,
             on_failure=fail,
             on_summary=lambda: result.detach(sync=False),
         )
@@ -550,12 +558,14 @@ def has_transaction(self):
     def _close_transaction(self):
         self._transaction = None
 
-    def begin_transaction(self, bookmark=None):
+    def begin_transaction(self, bookmark=None, metadata=None, timeout=None):
         """ Create a new :class:`.Transaction` within this session.
         Calling this method with a bookmark is equivalent to
 
         :param bookmark: a bookmark to which the server should
                          synchronise before beginning the transaction
+        :param metadata:
+        :param timeout:
         :returns: new :class:`.Transaction` instance.
         :raise: :class:`.TransactionError` if a transaction is already open
         """
@@ -572,13 +582,13 @@ def begin_transaction(self, bookmark=None):
                 _warned_about_transaction_bookmarks = True
             self._bookmarks_in = tuple([bookmark])
 
-        self._open_transaction()
+        self._open_transaction(metadata=metadata, timeout=timeout)
         return self._transaction
 
-    def _open_transaction(self, access_mode=None, metadata=None):
+    def _open_transaction(self, access_mode=None, metadata=None, timeout=None):
         self._transaction = Transaction(self, on_close=self._close_transaction)
         self._connect(access_mode)
-        self._connection.begin(self._bookmarks_in)
+        self._connection.begin(bookmarks=self._bookmarks_in, metadata=metadata, timeout=timeout)
 
     def commit_transaction(self):
         """ Commit the current transaction.

test/integration/test_session.py

@@ -19,14 +19,15 @@
 # limitations under the License.
 
 
+from time import sleep
 from unittest import SkipTest
 from uuid import uuid4
 
 from neo4j import \
     READ_ACCESS, WRITE_ACCESS, \
     CypherError, SessionError, TransactionError, unit_of_work
 from neo4j.types.graph import Node, Relationship, Path
-from neo4j.exceptions import CypherSyntaxError
+from neo4j.exceptions import CypherSyntaxError, TransientError
 
 from test.integration.tools import DirectIntegrationTestCase
 
@@ -62,6 +63,24 @@ def test_can_run_simple_statement_with_params(self):
         session.close()
         assert count == 1
 
+    def test_autocommit_transactions_use_bookmarks(self):
+        if self.protocol_version() < 3:
+            raise SkipTest("Test requires Bolt v3")
+        bookmarks = []
+        # Generate an initial bookmark
+        with self.driver.session() as session:
+            session.run("CREATE ()").consume()
+            bookmark = session.last_bookmark()
+            self.assertIsNotNone(bookmark)
+            bookmarks.append(bookmark)
+        # Propagate into another session
+        with self.driver.session(bookmarks=bookmarks) as session:
+            self.assertEqual(list(session.next_bookmarks()), bookmarks)
+            session.run("CREATE ()").consume()
+            bookmark = session.last_bookmark()
+            self.assertIsNotNone(bookmark)
+            self.assertNotIn(bookmark, bookmarks)
+
     def test_fails_on_bad_syntax(self):
         session = self.driver.session()
         with self.assertRaises(CypherError):
@@ -369,6 +388,27 @@ def test_last_run_statement_should_be_cleared_on_failure(self):
             assert connection_2._last_run_statement is None
             tx.close()
 
+    def test_transaction_metadata(self):
+        if self.protocol_version() < 3:
+            raise SkipTest("Test requires Bolt v3")
+        with self.driver.session() as session:
+            metadata_in = {"foo": "bar"}
+            with session.begin_transaction(metadata=metadata_in) as tx:
+                metadata_out = tx.run("CALL dbms.getTXMetaData").single().value()
+            self.assertEqual(metadata_in, metadata_out)
+
+    def test_transaction_timeout(self):
+        if self.protocol_version() < 3:
+            raise SkipTest("Test requires Bolt v3")
+        with self.driver.session() as s1:
+            s1.run("CREATE (a:Node)").consume()
+            with self.driver.session() as s2:
+                tx1 = s1.begin_transaction()
+                tx1.run("MATCH (a:Node) SET a.property = 1").consume()
+                tx2 = s2.begin_transaction(timeout=0.25)
+                with self.assertRaises(TransientError):
+                    tx2.run("MATCH (a:Node) SET a.property = 2").consume()
+
 
 class BookmarkingTestCase(DirectIntegrationTestCase):