Tx config for all tx types

Author: technige

neo4j/__init__.py

@@ -32,6 +32,7 @@
     "RoutingDriver",
     "Session",
     "Transaction",
+    "Statement",
     "StatementResult",
     "BoltStatementResult",
     "BoltStatementResultSummary",
@@ -72,7 +73,7 @@
 from warnings import warn
 
 
-from .compat import perf_counter, urlparse
+from .compat import perf_counter, urlparse, xstr
 from .config import *
 from .meta import version as __version__
 
@@ -436,30 +437,49 @@ def run(self, statement, parameters=None, **kwparameters):
         protocol_version = cx.protocol_version
         server = cx.server
 
-        statement = ustr(statement)
+        has_transaction = self.has_transaction()
+
+        statement_text = ustr(statement)
+        statement_metadata = getattr(statement, "metadata", None)
+        statement_timeout = getattr(statement, "timeout", None)
         parameters = fix_parameters(dict(parameters or {}, **kwparameters), protocol_version,
                                     supports_bytes=server.supports("bytes"))
 
         def fail(_):
             self._close_transaction()
 
         hydrant = PackStreamHydrator(protocol_version)
-        metadata = {
-            "statement": statement,
+        result_metadata = {
+            "statement": statement_text,
             "parameters": parameters,
             "server": server,
             "protocol_version": protocol_version,
         }
+        run_metadata = {
+            "metadata": statement_metadata,
+            "timeout": statement_timeout,
+            "on_success": result_metadata.update,
+            "on_failure": fail,
+        }
 
         def done(summary_metadata):
-            metadata.update(summary_metadata)
-            bookmark = metadata.get("bookmark")
+            result_metadata.update(summary_metadata)
+            bookmark = result_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)
+        self._last_result = result = BoltStatementResult(self, hydrant, result_metadata)
+
+        if has_transaction:
+            if statement_metadata:
+                raise ValueError("Metadata can only be attached at transaction level")
+            if statement_timeout:
+                raise ValueError("Timeouts only apply at transaction level")
+        else:
+            run_metadata["bookmarks"] = self._bookmarks_in
+
+        cx.run(statement_text, parameters, **run_metadata)
         cx.pull_all(
             on_records=lambda records: result._records.extend(
                 hydrant.hydrate_records(result.keys(), records)),
@@ -468,7 +488,7 @@ def done(summary_metadata):
             on_summary=lambda: result.detach(sync=False),
         )
 
-        if not self.has_transaction():
+        if not has_transaction:
             try:
                 self._connection.send()
                 self._connection.fetch()
@@ -805,6 +825,36 @@ def closed(self):
         return self._closed
 
 
+class Statement(object):
+
+    def __init__(self, text, metadata=None, timeout=None):
+        self.text = text
+        try:
+            self.metadata = metadata
+        except TypeError:
+            raise TypeError("Metadata must be coercible to a dict")
+        try:
+            self.timeout = timeout
+        except TypeError:
+            raise TypeError("Timeout must be specified as a number of seconds")
+
+    def __str__(self):
+        return xstr(self.text)
+
+
+def fix_parameters(parameters, protocol_version, **kwargs):
+    if not parameters:
+        return {}
+    dehydrator = PackStreamDehydrator(protocol_version, **kwargs)
+    try:
+        dehydrated, = dehydrator.dehydrate([parameters])
+    except TypeError as error:
+        value = error.args[0]
+        raise TypeError("Parameters of type {} are not supported".format(type(value).__name__))
+    else:
+        return dehydrated
+
+
 class StatementResult(object):
     """ A handler for the result of Cypher statement execution. Instances
     of this class are typically constructed and returned by
@@ -1406,19 +1456,6 @@ def custom_auth(principal, credentials, realm, scheme, **parameters):
     return AuthToken(scheme, principal, credentials, realm, **parameters)
 
 
-def fix_parameters(parameters, protocol_version, **kwargs):
-    if not parameters:
-        return {}
-    dehydrator = PackStreamDehydrator(protocol_version, **kwargs)
-    try:
-        dehydrated, = dehydrator.dehydrate([parameters])
-    except TypeError as error:
-        value = error.args[0]
-        raise TypeError("Parameters of type {} are not supported".format(type(value).__name__))
-    else:
-        return dehydrated
-
-
 def iter_items(iterable):
     """ Iterate through all items (key-value pairs) within an iterable
     dictionary-like object. If the object has a `keys` method, this is

requirements.txt

@@ -1,2 +1,2 @@
-neobolt==1.7.0rc3
+neobolt==1.7.0rc4
 neotime

setup.py

@@ -30,7 +30,7 @@
 from neo4j.meta import package, version
 
 install_requires = [
-    "neobolt==1.7.0rc3",
+    "neobolt==1.7.0rc4",
     "neotime",
 ]
 classifiers = [

test/integration/test_session.py

@@ -25,7 +25,7 @@
 
 from neo4j import \
     READ_ACCESS, WRITE_ACCESS, \
-    CypherError, SessionError, TransactionError, unit_of_work
+    CypherError, SessionError, TransactionError, unit_of_work, Statement
 from neo4j.types.graph import Node, Relationship, Path
 from neo4j.exceptions import CypherSyntaxError, TransientError
 
@@ -175,6 +175,32 @@ def test_should_not_allow_empty_statements(self):
             with self.assertRaises(ValueError):
                 _ = session.run("")
 
+    def test_statement_object(self):
+        if self.protocol_version() < 3:
+            raise SkipTest("Test requires Bolt v3")
+        with self.driver.session() as session:
+            value = session.run(Statement("RETURN $x"), x=1).single().value()
+            self.assertEqual(value, 1)
+
+    def test_autocommit_transactions_should_support_metadata(self):
+        if self.protocol_version() < 3:
+            raise SkipTest("Test requires Bolt v3")
+        metadata_in = {"foo": "bar"}
+        with self.driver.session() as session:
+            metadata_out = session.run(Statement("CALL dbms.getTXMetaData", metadata=metadata_in)).single().value()
+            self.assertEqual(metadata_in, metadata_out)
+
+    def test_autocommit_transactions_should_support_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()
+                with self.assertRaises(TransientError):
+                    s2.run(Statement("MATCH (a:Node) SET a.property = 2", timeout=0.25)).consume()
+
 
 class SummaryTestCase(DirectIntegrationTestCase):
 
@@ -388,6 +414,12 @@ def test_last_run_statement_should_be_cleared_on_failure(self):
             assert connection_2._last_run_statement is None
             tx.close()
 
+    def test_statement_object_not_supported(self):
+        with self.driver.session() as session:
+            with session.begin_transaction() as tx:
+                with self.assertRaises(ValueError):
+                    tx.run(Statement("RETURN 1", timeout=0.25))
+
     def test_transaction_metadata(self):
         if self.protocol_version() < 3:
             raise SkipTest("Test requires Bolt v3")

test/integration/test_types.py

@@ -25,9 +25,9 @@
 from pytz import FixedOffset, timezone, utc
 
 from neo4j.exceptions import CypherTypeError
-from neo4j.v1.types.graph import Node, Relationship, Path
-from neo4j.v1.types.spatial import CartesianPoint, WGS84Point
-from neo4j.v1.types.temporal import Duration, Date, Time, DateTime
+from neo4j.types.graph import Node, Relationship, Path
+from neo4j.types.spatial import CartesianPoint, WGS84Point
+from neo4j.types.temporal import Duration, Date, Time, DateTime
 
 from test.integration.tools import DirectIntegrationTestCase
 

test/stub/scripts/bookmark_chain_with_autocommit.script

@@ -1,25 +1,19 @@
-!: AUTO INIT
+!: BOLT 3
+!: AUTO HELLO
+!: AUTO GOODBYE
 !: AUTO RESET
 
-C: RUN "BEGIN" {"bookmark": "bookmark:1", "bookmarks": ["bookmark:1"]}
-   DISCARD_ALL
+C: BEGIN {"bookmarks": ["bookmark:1"]}
 S: SUCCESS {}
-   SUCCESS {}
-C: RUN "COMMIT" {}
-   DISCARD_ALL
-S: SUCCESS {"bookmark": "bookmark:2", "bookmarks": ["bookmark:2"]}
-   SUCCESS {}
+C: COMMIT
+S: SUCCESS {"bookmark": "bookmark:2"}
 
-C: RUN "RETURN 1" {}
+C: RUN "RETURN 1" {} {"bookmarks": ["bookmark:2"]}
    PULL_ALL
-S: SUCCESS {"bookmark": "bookmark:x", "bookmarks": ["bookmark:x"]}
+S: SUCCESS {"bookmark": "bookmark:3"}
    SUCCESS {}
 
-C: RUN "BEGIN" {"bookmark": "bookmark:2", "bookmarks": ["bookmark:2"]}
-   DISCARD_ALL
+C: BEGIN {"bookmarks": ["bookmark:3"]}
 S: SUCCESS {}
-   SUCCESS {}
-C: RUN "COMMIT" {}
-   DISCARD_ALL
-S: SUCCESS {"bookmark": "bookmark:3", "bookmarks": ["bookmark:3"]}
-   SUCCESS {}
+C: COMMIT
+S: SUCCESS {"bookmark": "bookmark:4"}

test/stub/test_bookmarking.py

@@ -59,7 +59,7 @@ def test_should_automatically_chain_bookmarks(self):
                         pass
                     assert session.last_bookmark() == "bookmark:3"
 
-    def test_autocommit_transaction_does_not_break_chain(self):
+    def test_autocommit_transaction_included_in_chain(self):
         with StubCluster({9001: "router.script", 9004: "bookmark_chain_with_autocommit.script"}):
             uri = "bolt+routing://localhost:9001"
             with GraphDatabase.driver(uri, auth=self.auth_token, encrypted=False) as driver:
@@ -68,7 +68,7 @@ def test_autocommit_transaction_does_not_break_chain(self):
                         pass
                     assert session.last_bookmark() == "bookmark:2"
                     session.run("RETURN 1").consume()
-                    assert session.last_bookmark() == "bookmark:2"
+                    assert session.last_bookmark() == "bookmark:3"
                     with session.begin_transaction():
                         pass
-                    assert session.last_bookmark() == "bookmark:3"
+                    assert session.last_bookmark() == "bookmark:4"

test/unit/test_types.py

@@ -22,8 +22,8 @@
 from unittest import TestCase
 
 from neobolt.packstream import Structure
-from neo4j.v1.types import PackStreamHydrator
-from neo4j.v1.types.graph import Node, Path, Graph
+from neo4j.types import PackStreamHydrator
+from neo4j.types.graph import Node, Path, Graph
 
 
 class NodeTestCase(TestCase):