Introduce API Redesign

neo4j/_async/driver.py

@@ -33,6 +33,7 @@
 )
 from ..addressing import Address
 from ..api import (
+    CLUSTER_AUTO_ACCESS,
     READ_ACCESS,
     TRUST_ALL_CERTIFICATES,
     TRUST_SYSTEM_CA_SIGNED_CERTIFICATES,
@@ -392,6 +393,155 @@ async def supports_multi_db(self):
             await session._connect(READ_ACCESS)
             return session._connection.supports_multiple_databases
 
+    async def query(
+        self, query, parameters=None, database=None,
+        cluster_member_access=CLUSTER_AUTO_ACCESS, skip_records=False,
+        metadata=None, timeout=None, **kwargs
+    ):
+        """
+        Run a Cypher query within an managed transaction.
+
+        The query is sent and the result header received
+        immediately and the :class:`neo4j.QueryResult`is
+        fetched.
+
+        For more usage details, see :meth:`.AsyncTransaction.query`.
+
+        For auto-commit queries, use :meth:`AsyncSession.run`.
+
+        For access to the :class:`neo4j.AsyncResult` object,
+        use :meth:`AsyncDriver.execute` and :meth:`.AsyncTransaction.run`
+
+        :param query: cypher query
+        :type query: str, neo4j.Query
+
+        :param parameters: dictionary of parameters
+        :type parameters: dict
+
+        :param database: the name of the database to be used
+        :type database: str
+
+        :param cluster_member_access: the kind of cluster member used
+            for running the work
+
+        :param metadata:
+            a dictionary with metadata.
+            For more usage details,
+            see :meth:`.AsyncSession.begin_transaction`.
+        :type metadata: dict
+
+        :param timeout:
+            the transaction timeout in seconds.
+            For more usage details,
+            see :meth:`.AsyncSession.begin_transaction`.
+        :type timeout: int
+
+        :param kwargs: additional keyword parameters
+
+        :returns: a new :class:`neo4j.QueryResult` object
+        :rtype: QueryResult
+        """
+
+        async with self.session(database=database) as session:
+            return await session.query(
+                query,
+                parameters=parameters,
+                cluster_member_access=cluster_member_access,
+                skip_records=skip_records,
+                timeout=timeout,
+                metadata=metadata,
+                **kwargs
+            )
+
+    async def execute(
+        self, transaction_function,
+        database=None, cluster_member_access=CLUSTER_AUTO_ACCESS,
+        metadata=None, timeout=None
+    ):
+        """Execute a unit of work in a managed transaction.
+
+        This transaction will automatically be committed unless an exception
+        is thrown during query execution or by the user code.
+        Note, that this function perform retries and that the supplied
+        ``transaction_function`` might get invoked more than once.
+
+        Example::
+
+            async def do_cypher_tx(tx, cypher):
+                records, _ = await tx.query(cypher)
+                return records
+
+            values = await driver.execute(lambda tx: do_cypher_tx(tx, "RETURN 1 AS x"))
+
+        Example::
+
+            async def do_cypher_tx(tx):
+                records, _ = await tx.query("RETURN 1 AS x")
+                return records
+
+            values = await driver.execute(
+                do_cypher_tx,
+                database="neo4j",
+                cluster_member_access=neo4j.api.CLUSTER_READERS_ACCESS
+            )
+
+        Example::
+
+            async def get_two_tx(tx):
+                result = await tx.run("UNWIND [1,2,3,4] AS x RETURN x")
+                values = []
+                async for record in result:
+                    if len(values) >= 2:
+                        break
+                    values.append(record.values())
+                # or shorter: values = [record.values()
+                #                       for record in await result.fetch(2)]
+
+                # discard the remaining records if there are any
+                summary = await result.consume()
+                # use the summary for logging etc.
+                return values
+
+
+            values = await driver.execute(get_two_tx)
+
+        :param transaction_function: a function that takes a transaction as an
+            argument and does work with the transaction.
+            ``transaction_function(tx)`` where ``tx`` is a
+            :class:`.Transaction`.
+
+        :param parameters: dictionary of parameters
+        :type parameters: dict
+
+        :param database: the name of the database to be used
+        :type database: str
+
+        :param cluster_member_access: the kind of cluster member used
+            for running the work
+
+        :param metadata:
+            a dictionary with metadata.
+            For more usage details,
+            see :meth:`.AsyncSession.begin_transaction`.
+        :type metadata: dict
+
+        :param timeout:
+            the transaction timeout in seconds.
+            For more usage details,
+            see :meth:`.AsyncSession.begin_transaction`.
+        :type timeout: int
+
+        :return: a result as returned by the given unit of work
+        """
+
+        async with self.session(database=database) as session:
+            return await session.execute(
+                transaction_function,
+                cluster_member_access=cluster_member_access,
+                timeout=timeout,
+                metadata=metadata
+            )
+
 
 class AsyncBoltDriver(_Direct, AsyncDriver):
     """:class:`.AsyncBoltDriver` is instantiated for ``bolt`` URIs and

neo4j/_async/io/_pool.py

@@ -348,6 +348,9 @@ async def acquire(
             self.address, deadline, liveness_check_timeout
         )
 
+    def is_direct(self):
+        return True
+
 
 class AsyncNeo4jPool(AsyncIOPool):
     """ Connection pool with routing table.
@@ -789,3 +792,6 @@ def on_write_failure(self, address):
         for database in self.routing_tables.keys():
             self.routing_tables[database].writers.discard(address)
         log.debug("[#0000]  C: <ROUTING> table=%r", self.routing_tables)
+
+    def is_direct(self):
+        return False

neo4j/_async/work/result.py

@@ -18,6 +18,7 @@
 
 from collections import deque
 from warnings import warn
+from collections import namedtuple
 
 from ..._async_compat.util import AsyncUtil
 from ..._codec.hydration import BrokenHydrationObject
@@ -707,3 +708,5 @@ def closed(self):
         .. versionadded:: 5.0
         """
         return self._out_of_scope or self._consumed
+
+QueryResult = namedtuple("QueryResult", ("records", "summary"))