Use supported event loop behavior

jupyter_core/application.py

@@ -76,6 +76,9 @@ class JupyterApp(Application):
     aliases = base_aliases
     flags = base_flags
 
+    # Set to True for tornado-based apps.
+    _prefer_selector_loop = False
+
     def _log_level_default(self) -> int:
         return logging.INFO
 
@@ -302,7 +305,7 @@ def launch_instance(cls, argv: t.Any = None, **kwargs: t.Any) -> None:
 
         If a global instance already exists, this reinitializes and starts it
         """
-        loop = get_event_loop()
+        loop = get_event_loop(cls._prefer_selector_loop)
         coro = cls._async_launch_instance(argv, **kwargs)
         loop.run_until_complete(coro)
 

jupyter_core/utils/__init__.py

@@ -9,6 +9,7 @@
 import sys
 import threading
 import warnings
+from contextvars import ContextVar
 from pathlib import Path
 from types import FrameType
 from typing import Any, Awaitable, Callable, TypeVar, cast
@@ -126,6 +127,7 @@ def run(self, coro: Any) -> Any:
 
 
 _runner_map: dict[str, _TaskRunner] = {}
+_loop: ContextVar[asyncio.AbstractEventLoop | None] = ContextVar("_loop", default=None)
 
 
 def run_sync(coro: Callable[..., Awaitable[T]]) -> Callable[..., T]:
@@ -186,19 +188,18 @@ async def ensure_async(obj: Awaitable[T] | T) -> T:
     return cast(T, obj)
 
 
-def get_event_loop() -> asyncio.AbstractEventLoop:
-    # Get the loop for this thread.
-    # In Python 3.12, a deprecation warning is raised, which
-    # may later turn into a RuntimeError.  We handle both
-    # cases.
-    with warnings.catch_warnings():
-        warnings.simplefilter("ignore", DeprecationWarning)
-        try:
-            loop = asyncio.get_event_loop()
-        except RuntimeError:
-            if sys.platform == "win32":
-                loop = asyncio.WindowsSelectorEventLoopPolicy().new_event_loop()
-            else:
-                loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
+def get_event_loop(prefer_selector_loop: bool = False) -> asyncio.AbstractEventLoop:
+    # Get the loop for this thread, or create a new one.
+    loop = _loop.get()
+    if loop is not None and not loop.is_closed():
+        return loop
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        if sys.platform == "win32" and prefer_selector_loop:
+            loop = asyncio.WindowsSelectorEventLoopPolicy().new_event_loop()
+        else:
+            loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+    _loop.set(loop)
     return loop

tests/test_application.py

@@ -139,3 +139,13 @@ def test_async_app():
     AsyncApp.launch_instance([])
     app = AsyncApp.instance()
     assert app.value == 10
+
+
+class AsyncTornadoApp(AsyncApp):
+    _prefer_selector_loop = True
+
+
+def test_async_tornado_app():
+    AsyncApp.launch_instance([])
+    app = AsyncApp.instance()
+    assert app.value == 10

tests/test_utils.py

@@ -10,7 +10,13 @@
 
 import pytest
 
-from jupyter_core.utils import deprecation, ensure_async, ensure_dir_exists, run_sync
+from jupyter_core.utils import (
+    deprecation,
+    ensure_async,
+    ensure_dir_exists,
+    get_event_loop,
+    run_sync,
+)
 
 
 def test_ensure_dir_exists():
@@ -42,11 +48,11 @@ async def foo():
     foo_sync = run_sync(foo)
     assert foo_sync() == 1
     assert foo_sync() == 1
-    asyncio.get_event_loop().close()
+    get_event_loop().close()
 
     asyncio.set_event_loop(None)
     assert foo_sync() == 1
-    asyncio.get_event_loop().close()
+    get_event_loop().close()
 
     asyncio.run(foo())