Introduce Driver.execute_query

nutkit/frontend/driver.py

@@ -67,6 +67,26 @@ def send_and_receive(self, req, timeout=None, hooks=None, *,
         return self.receive(timeout=timeout, hooks=hooks,
                             allow_resolution=allow_resolution)
 
+    def execute_query(self, cypher, params=None,
+                      routing=None, database=None,
+                      impersonated_user=None, bookmark_manager=...):
+        config = {
+            "routing": routing,
+            "database": database,
+            "impersonatedUser": impersonated_user
+        }
+
+        if bookmark_manager is None:
+            config["bookmarkManagerId"] = -1
+        elif bookmark_manager is not Ellipsis:
+            config["bookmarkManagerId"] = bookmark_manager.id
+
+        req = protocol.ExecuteQuery(self._driver.id, cypher, params, config)
+        res = self.send_and_receive(req, allow_resolution=True)
+        if not isinstance(res, protocol.EagerResult):
+            raise Exception("Should be EagerResult but was: %s" % res)
+        return res
+
     def verify_connectivity(self):
         req = protocol.VerifyConnectivity(self._driver.id)
         res = self.send_and_receive(req, allow_resolution=True)

nutkit/protocol/feature.py

@@ -13,6 +13,9 @@ class Feature(Enum):
     # is picked up instead or we need to wait until the full pool depletes.
     API_CONNECTION_ACQUISITION_TIMEOUT = \
         "Feature:API:ConnectionAcquisitionTimeout"
+    # The driver offers a method to run a query in a retryable context at the
+    # driver object level.
+    API_DRIVER_EXECUTE_QUERY = "Feature:API:Driver.ExecuteQuery"
     # The driver offers a method for checking if a connection to the remote
     # server of cluster can be established and retrieve the server info of the
     # reached remote.

nutkit/protocol/requests.py

@@ -592,3 +592,30 @@ def __init__(self, result_id, record_key, type_name, field_id):
         self.recordKey = record_key
         self.type = type_name
         self.field = field_id
+
+
+class ExecuteQuery:
+    """
+    Request to execute a query in a retriable context.
+
+    Backend should return EagerResult or a Error response.
+
+    :param driver_id: The id of the driver where the cypher query has to run.
+    :param cypher: The cypher query which to run.
+    :param params: The cypher query params.
+    :param config: The configuration
+    :param config.database: The database where the query will run.
+    :param config.routing: The type of routing ("w" for Writers,
+         "r" for "Readers")
+    :param config.impersonatedUser: The user which will be impersonated
+    :param config.bookmarkManagerId: The id of the bookmark manager
+        used in the query. None or not define for using the default,
+        -1 for disabling the BookmarkManager
+    """
+
+    def __init__(self, driver_id, cypher, params, config):
+        self.driverId = driver_id
+        self.cypher = cypher
+        self.params = params
+        if config:
+            self.config = config

nutkit/protocol/responses.py

@@ -495,6 +495,13 @@ def __init__(self, inUse, idle):
         self.idle = idle
 
 
+class EagerResult:
+    def __init__(self, keys, records, summary):
+        self.keys = keys
+        self.records = [Record(**record) for record in records or []]
+        self.summary = Summary(**summary)
+
+
 class BaseError(Exception):
     """
     Base class for all types of errors, should not be sent from backend.

tests/stub/driver_execute_query/scripts/router.script

@@ -0,0 +1,10 @@
+!: BOLT 5.0
+!: ALLOW RESTART
+
+A: HELLO {"{}": "*"}
+{+
+    C: ROUTE "*" "*" "*"
+    S: SUCCESS {"rt": {"ttl": 1000, "db": "adb", "servers": [{"addresses": ["#HOST#:9000"], "role":"ROUTE"}, {"addresses": ["#HOST#:9010"], "role":"READ"}, {"addresses": ["#HOST#:9020"], "role":"WRITE"}]}}
+    ?: RESET
++}
+?: GOODBYE

tests/stub/driver_execute_query/scripts/router_invert_reader_and_writer_second_call.script

@@ -0,0 +1,12 @@
+!: BOLT 5.0
+!: ALLOW RESTART
+
+A: HELLO {"{}": "*"}
+C: ROUTE "*" "*" "*"
+S: SUCCESS {"rt": {"ttl": 1000, "db": "adb", "servers": [{"addresses": ["#HOST#:9000"], "role":"ROUTE"}, {"addresses": ["#HOST#:9010"], "role":"READ"}, {"addresses": ["#HOST#:9020"], "role":"WRITE"}]}}
+?: RESET
+C: ROUTE "*" "*" "*"
+S: SUCCESS {"rt": {"ttl": 1000, "db": "adb", "servers": [{"addresses": ["#HOST#:9000"], "role":"ROUTE"}, {"addresses": ["#HOST#:9020"], "role":"READ"}, {"addresses": ["#HOST#:9010"], "role":"WRITE"}]}}
+?: RESET
+
+?: GOODBYE

tests/stub/driver_execute_query/scripts/router_with_db_name.script

@@ -0,0 +1,8 @@
+!: BOLT 5.0
+!: ALLOW RESTART
+
+A: HELLO {"{}": "*"}
+C: ROUTE "*" "*" {"db":"neo4j"}
+S: SUCCESS {"rt": {"ttl": 1000, "db": "neo4j", "servers": [{"addresses": ["#HOST#:9000"], "role":"ROUTE"}, {"addresses": ["#HOST#:9010"], "role":"READ"}, {"addresses": ["#HOST#:9020"], "role":"WRITE"}]}}
+*: RESET
+?: GOODBYE

tests/stub/driver_execute_query/scripts/router_with_impersonation.script

@@ -0,0 +1,8 @@
+!: BOLT 5.0
+!: ALLOW RESTART
+
+A: HELLO {"{}": "*"}
+C: ROUTE "*" "*" {"imp_user": "that-other-dude"}
+S: SUCCESS {"rt": {"ttl": 1000, "db": "adb", "servers": [{"addresses": ["#HOST#:9000"], "role":"ROUTE"}, {"addresses": ["#HOST#:9010"], "role":"READ"}, {"addresses": ["#HOST#:9020"], "role":"WRITE"}]}}
+*: RESET
+?: GOODBYE

tests/stub/driver_execute_query/scripts/transaction_chaining.script

@@ -0,0 +1,39 @@
+!: BOLT 5.0
+
+A: HELLO {"{}": "*"}
+*: RESET
+
+C: BEGIN {"[db]": "*"}
+S: SUCCESS {}
+C: RUN "CREATE (p:Person {name:$name}) RETURN p.name AS name" {"name": "the person"} {}
+S: SUCCESS {"fields": ["name"]}
+C: PULL {"n": 1000}
+S: RECORD ["the person"]
+   SUCCESS {"type": "w"}
+C: COMMIT
+S: SUCCESS {"bookmark": "bm1"}
+*: RESET
+
+{{
+    C: BEGIN {"[db]": "*"}
+    S: SUCCESS {}
+    C: RUN "MATCH (p:Person {name:$name}) RETURN p.name AS name" {"name": "the person"} {}
+    S: SUCCESS {"fields": ["name"]}
+    C: PULL {"n": 1000}
+    S: SUCCESS {"type": "r"}
+    C: COMMIT
+    S: SUCCESS {"bookmark": "bm1"}
+----
+    C: BEGIN {"bookmarks": ["bm1"], "[db]": "*"}
+    S: SUCCESS {}
+    C: RUN "MATCH (p:Person {name:$name}) RETURN p.name AS name" {"name": "the person"} {}
+    S: SUCCESS {"fields": ["name"]}
+    C: PULL {"n": 1000}
+    S: RECORD ["the person"]
+       SUCCESS {"type": "r"}
+    C: COMMIT
+    S: SUCCESS {"bookmark": "bm2"}
+}}
+*: RESET
+
+?: GOODBYE

tests/stub/driver_execute_query/scripts/transaction_chaining_custom_bmm.script

@@ -0,0 +1,39 @@
+!: BOLT 5.0
+
+A: HELLO {"{}": "*"}
+*: RESET
+
+C: BEGIN {"bookmarks": ["other_db:bm1"], "[db]": "*"}
+S: SUCCESS {}
+C: RUN "CREATE (p:Person {name:$name}) RETURN p.name AS name" {"name": "a person"} {}
+S: SUCCESS {"fields": ["name"]}
+C: PULL {"n": 1000}
+S: RECORD ["a person"]
+   SUCCESS {"type": "w"}
+C: COMMIT
+S: SUCCESS {"bookmark": "bm1"}
+*: RESET
+
+{{
+    C: BEGIN {"bookmarks": ["other_db:bm1"], "[db]": "*"}
+    S: SUCCESS {}
+    C: RUN "MATCH (p:Person {name:$name}) RETURN p.name AS name" {"name": "a person"} {}
+    S: SUCCESS {"fields": ["name"]}
+    C: PULL {"n": 1000}
+    S: SUCCESS {"type": "r"}
+    C: COMMIT
+    S: SUCCESS {"bookmark": "bm1"}
+----
+    C: BEGIN {"bookmarks{}": ["bm1", "other_db:bm1"], "[db]": "*"}
+    S: SUCCESS {}
+    C: RUN "MATCH (p:Person {name:$name}) RETURN p.name AS name" {"name": "a person"} {}
+    S: SUCCESS {"fields": ["name"]}
+    C: PULL {"n": 1000}
+    S: RECORD ["a person"]
+       SUCCESS {"type": "r"}
+    C: COMMIT
+    S: SUCCESS {"bookmark": "bm2"}
+}}
+*: RESET
+
+?: GOODBYE

tests/stub/driver_execute_query/scripts/tx_return_1.script

@@ -0,0 +1,15 @@
+!: BOLT 5.0
+
+A: HELLO {"{}": "*"}
+*: RESET
+C: BEGIN {"{}": "*"}
+S: SUCCESS {}
+C: RUN "RETURN 1 AS n" {} {}
+S: SUCCESS {"fields": ["n"], "qid": 1}
+C: PULL {"n": 1000}
+S: RECORD [1]
+   SUCCESS {"type": "r"}
+C: COMMIT
+S: SUCCESS {"bookmark": "neo4j:bookmark:v1:tx424242"}
+*: RESET
+?: GOODBYE

tests/stub/driver_execute_query/scripts/tx_return_1_disconnect_on_pull.script

@@ -0,0 +1,10 @@
+!: BOLT 5.0
+
+A: HELLO {"{}": "*"}
+*: RESET
+C: BEGIN {"{}": "*"}
+S: SUCCESS {}
+C: RUN "RETURN 1 AS n" {} {}
+S: SUCCESS {"fields": ["n"], "qid": 1}
+C: PULL {"n": 1000}
+S: <EXIT>

tests/stub/driver_execute_query/scripts/tx_return_1_transaction_terminated_on_pull.script

@@ -0,0 +1,12 @@
+!: BOLT 5.0
+
+A: HELLO {"{}": "*"}
+*: RESET
+C: BEGIN {"{}": "*"}
+S: SUCCESS {}
+C: RUN "RETURN 1 AS n" {} {}
+S: SUCCESS {"fields": ["n"], "qid": 1}
+C: PULL {"n": 1000}
+S: FAILURE {"code": "Neo.ClientError.Transaction.Terminated", "message": "message"}
+*: RESET
+?: GOODBYE

tests/stub/driver_execute_query/scripts/tx_return_1_with_impersonation.script

@@ -0,0 +1,15 @@
+!: BOLT 5.0
+
+A: HELLO {"{}": "*"}
+*: RESET
+C: BEGIN {"imp_user": "that-other-dude", "db": "adb"}
+S: SUCCESS {}
+C: RUN "RETURN 1 AS n" {} {}
+S: SUCCESS {"fields": ["n"], "qid": 1}
+C: PULL {"n": 1000}
+S: RECORD [1]
+   SUCCESS {"type": "r"}
+C: COMMIT
+S: SUCCESS {"bookmark": "neo4j:bookmark:v1:tx424242"}
+*: RESET
+?: GOODBYE

tests/stub/driver_execute_query/scripts/tx_return_1_with_params.script

@@ -0,0 +1,15 @@
+!: BOLT 5.0
+
+A: HELLO {"{}": "*"}
+*: RESET
+C: BEGIN {"{}": "*"}
+S: SUCCESS {}
+C: RUN "RETURN $a AS n" { "a": 1 } {}
+S: SUCCESS {"fields": ["n"], "qid": 1}
+C: PULL {"n": 1000}
+S: RECORD [1]
+   SUCCESS {"type": "r", "db": "#DB#"}
+C: COMMIT
+S: SUCCESS {"bookmark": "neo4j:bookmark:v1:tx424242"}
+*: RESET
+?: GOODBYE