Update AyncCondition to be based on Python 3.11 (#879)

Most notably, this changes how the async driver behaves when created while no
event loop is running. Previously, this would lead to hard to predict errors 
only occurring once the connection pool ran full.

Now, the async synchronization primitives will defer binding to an event loop
until they're used in an async context for the first time. This simply allows
users to safely create a driver without a running event loop (in a sync 
context) and later use it in an async context.

Important note: this will likely only work when the user is on Python 3.10+
because the driver also relies on synchronization primitives that come with
`asyncio`. So their behavior depends on the used Python version.

+ Add test for creating async driver in sync environment

Author: robsdedude

src/neo4j/_async_compat/concurrency.py

@@ -305,26 +305,17 @@ class AsyncCondition:
     A new Lock object is created and used as the underlying lock.
     """
 
-    # copied and modified from Python 3.7's asyncio.locks module
-    # to add support for `.wait(timeout)`
+    # copied and modified from Python 3.11's asyncio package
+    # to add support for `.wait(timeout)` and cooperative locks
 
     # Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
     # 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020, 2021, 2022
     # Python Software Foundation;
     # All Rights Reserved
 
-    def __init__(self, lock=None, *, loop=None):
-        if loop is not None:
-            self._loop = loop
-        else:
-            self._loop = asyncio.get_event_loop()
-
+    def __init__(self, lock=None):
         if lock is None:
-            lock = asyncio.Lock(loop=self._loop)
-        elif (hasattr(lock, "_loop")
-              and lock._loop is not None
-              and lock._loop is not self._loop):
-            raise ValueError("loop argument must agree with lock")
+            lock = AsyncLock()
 
         self._lock = lock
         # Export the lock's locked(), acquire() and release() methods.
@@ -334,6 +325,23 @@ def __init__(self, lock=None, *, loop=None):
 
         self._waiters = collections.deque()
 
+    _loop = None
+    _loop_lock = threading.Lock()
+
+    def _get_loop(self):
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+
+        if self._loop is None:
+            with self._loop_lock:
+                if self._loop is None:
+                    self._loop = loop
+        if loop is not self._loop:
+            raise RuntimeError(f'{self!r} is bound to a different event loop')
+        return loop
+
     async def __aenter__(self):
         if isinstance(self._lock, (AsyncCooperativeLock,
                                    AsyncCooperativeRLock)):
@@ -374,7 +382,7 @@ async def _wait(self, timeout=None, me=None):
         else:
             self._lock.release()
         try:
-            fut = self._loop.create_future()
+            fut = self._get_loop().create_future()
             self._waiters.append(fut)
             try:
                 await wait_for(fut, timeout)
@@ -409,6 +417,19 @@ async def wait(self, timeout=None):
         me = asyncio.current_task()
         return await self._wait(timeout=timeout, me=me)
 
+    async def wait_for(self, predicate):
+        """Wait until a predicate becomes true.
+
+        The predicate should be a callable which result will be
+        interpreted as a boolean value.  The final predicate value is
+        the return value.
+        """
+        result = predicate()
+        while not result:
+            await self.wait()
+            result = predicate()
+        return result
+
     def notify(self, n=1):
         """By default, wake up one coroutine waiting on this condition, if any.
         If the calling coroutine has not acquired the lock when this method

tests/integration/mixed/test_async_driver.py

@@ -0,0 +1,94 @@
+# Copyright (c) "Neo4j"
+# Neo4j Sweden AB [https://neo4j.com]
+#
+# This file is part of Neo4j.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import asyncio
+import sys
+
+import pytest
+
+import neo4j
+
+from ... import env
+
+
+# TODO: Python 3.9: when support gets dropped, remove this mark
+@pytest.mark.xfail(
+    # direct driver is not making use of `asyncio.Lock`.
+    sys.version_info < (3, 10) and env.NEO4J_SCHEME == "neo4j",
+    reason="asyncio's synchronization primitives can create a new event loop "
+           "if instantiated while there is no running event loop. This "
+           "changed with Python 3.10.",
+    raises=RuntimeError,
+    strict=True,
+)
+def test_can_create_async_driver_outside_of_loop(uri, auth):
+    pool_size = 2
+    # used to make sure the pool was full at least at some point
+    counter = 0
+    was_full = False
+
+    async def return_1(tx):
+        nonlocal counter, was_full
+        res = await tx.run("RETURN 1")
+
+        counter += 1
+        while not was_full and counter < pool_size:
+            await asyncio.sleep(0.001)
+        if not was_full:
+            # a little extra time to make sure a connection too many was
+            # tried to be acquired from the pool
+            was_full = True
+            await asyncio.sleep(0.5)
+
+        await res.consume()
+        counter -= 1
+
+    async def run(driver: neo4j.AsyncDriver):
+        async with driver:
+            sessions = []
+            try:
+                for i in range(pool_size * 4):
+                    sessions.append(driver.session())
+                work_loads = (session.execute_read(return_1)
+                              for session in sessions)
+                await asyncio.gather(*work_loads)
+            finally:
+                cancelled = None
+                for session in sessions:
+                    if not cancelled:
+                        try:
+                            await session.close()
+                        except asyncio.CancelledError as e:
+                            session.cancel()
+                            cancelled = e
+                    else:
+                        session.cancel()
+                await driver.close()
+                if cancelled:
+                    raise cancelled
+
+
+    driver = neo4j.AsyncGraphDatabase.driver(
+        uri, auth=auth, max_connection_pool_size=pool_size
+    )
+    coro = run(driver)
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(coro)
+    finally:
+        loop.close()