Remove nest-asyncio

nbclient/util.py

@@ -4,72 +4,54 @@
 # Distributed under the terms of the Modified BSD License.
 
 import asyncio
+import atexit
 import inspect
-import sys
+import threading
 from typing import Any, Awaitable, Callable, Optional, TypeVar, Union
 
+T = TypeVar("T")
 
-def check_ipython() -> None:
-    # original from vaex/asyncio.py
-    IPython = sys.modules.get('IPython')
-    if IPython:
-        version_str = IPython.__version__
-        # We get rid of any trailing ".dev"
-        version_str = version_str.replace(".dev", "")
-
-        IPython_version = tuple(map(int, version_str.split('.')))
-        if IPython_version < (7, 0, 0):
-            raise RuntimeError(
-                f'You are using IPython {IPython.__version__} '
-                'while we require 7.0.0+, please update IPython'
-            )
-
-
-def check_patch_tornado() -> None:
-    """If tornado is imported, add the patched asyncio.Future to its tuple of acceptable Futures"""
-    # original from vaex/asyncio.py
-    if 'tornado' in sys.modules:
-        import tornado.concurrent
-
-        if asyncio.Future not in tornado.concurrent.FUTURES:
-            tornado.concurrent.FUTURES = tornado.concurrent.FUTURES + (  # type: ignore
-                asyncio.Future,
-            )
-
-
-def just_run(coro: Awaitable) -> Any:
-    """Make the coroutine run, even if there is an event loop running (using nest_asyncio)"""
-    try:
-        loop = asyncio.get_running_loop()
-    except RuntimeError:
-        loop = None
-    if loop is None:
-        had_running_loop = False
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-    else:
-        had_running_loop = True
-    if had_running_loop:
-        # if there is a running loop, we patch using nest_asyncio
-        # to have reentrant event loops
-        check_ipython()
-        import nest_asyncio
-
-        nest_asyncio.apply()
-        check_patch_tornado()
-    return loop.run_until_complete(coro)
 
+class _TaskRunner:
+    """A task runner that runs an asyncio event loop on a background thread."""
 
-T = TypeVar("T")
+    def __init__(self):
+        self.__io_loop: Optional[asyncio.AbstractEventLoop] = None
+        self.__runner_thread: Optional[threading.Thread] = None
+        self.__lock = threading.Lock()
+        atexit.register(self._close)
+
+    def _close(self):
+        if self.__io_loop:
+            self.__io_loop.stop()
+
+    def _runner(self):
+        loop = self.__io_loop
+        assert loop is not None
+        try:
+            loop.run_forever()
+        finally:
+            loop.close()
+
+    def run(self, coro):
+        """Synchronously run a coroutine on a background thread."""
+        with self.__lock:
+            name = f"{threading.current_thread().name} - runner"
+            if self.__io_loop is None:
+                self.__io_loop = asyncio.new_event_loop()
+                self.__runner_thread = threading.Thread(target=self._runner, daemon=True, name=name)
+                self.__runner_thread.start()
+        fut = asyncio.run_coroutine_threadsafe(coro, self.__io_loop)
+        return fut.result(None)
+
+
+_runner_map = {}
+_loop_map = {}
 
 
 def run_sync(coro: Callable[..., Awaitable[T]]) -> Callable[..., T]:
     """Runs a coroutine and blocks until it has executed.
 
-    An event loop is created if no one already exists. If an event loop is
-    already running, this event loop execution is nested into the already
-    running one if `nest_asyncio` is set to True.
-
     Parameters
     ----------
     coro : coroutine
@@ -82,7 +64,23 @@ def run_sync(coro: Callable[..., Awaitable[T]]) -> Callable[..., T]:
     """
 
     def wrapped(*args, **kwargs):
-        return just_run(coro(*args, **kwargs))
+        name = threading.current_thread().name
+        inner = coro(*args, **kwargs)
+        try:
+            # If a loop is currently running in this thread,
+            # use a task runner.
+            asyncio.get_running_loop()
+            if name not in _runner_map:
+                _runner_map[name] = _TaskRunner()
+            return _runner_map[name].run(inner)
+        except RuntimeError:
+            pass
+
+        # Run the loop for this thread.
+        if name not in _loop_map:
+            _loop_map[name] = asyncio.new_event_loop()
+        loop = _loop_map[name]
+        return loop.run_until_complete(inner)
 
     wrapped.__doc__ = coro.__doc__
     return wrapped

pyproject.toml

@@ -56,7 +56,6 @@ module = [
     "jupyter_core.*",
     "nbformat.*",
     "nbconvert.*",
-    "nest_asyncio.*",
     "testpath",
     "traitlets.*",
     "xmltodict"

requirements.txt

@@ -1,4 +1,3 @@
 jupyter_client>=6.1.5,<8
 nbformat>=5.0,<6
-nest_asyncio>=1.5.6,<2
 traitlets>=5.2.2,<6