Apply nest_asyncio for IPython with asyncio reactor

scrapy/utils/console.py

@@ -3,13 +3,17 @@ from __future__ import annotations
 import code
 from collections.abc import Callable
 from functools import wraps
+from logging import getLogger
 from typing import TYPE_CHECKING, Any
 
+from scrapy.utils.reactor import is_asyncio_reactor_installed
+
 if TYPE_CHECKING:
     from collections.abc import Iterable
 
 EmbedFuncT = Callable[..., None]
 KnownShellsT = dict[str, Callable[..., EmbedFuncT]]
+logger = getLogger(__name__)
 
 
 def _embed_ipython_shell(
@@ -30,6 +34,20 @@ def _embed_ipython_shell(
     @wraps(_embed_ipython_shell)
     def wrapper(namespace: dict[str, Any] = namespace or {}, banner: str = "") -> None:
         config = load_default_config()  # type: ignore[no-untyped-call]
+        try:
+            asyncio_reactor = is_asyncio_reactor_installed()
+        except RuntimeError:
+            asyncio_reactor = False
+        if asyncio_reactor:
+            try:
+                import nest_asyncio  # noqa: PLC0415
+            except ImportError:
+                logger.warning(
+                    "Install nest-asyncio to use the IPython shell with the "
+                    "asyncio reactor."
+                )
+                return
+            nest_asyncio.apply()
         # Always use .instance() to ensure _instance propagation to all parents
         # this is needed for <TAB> completion works well for new imports
         # and clear the instance to always have the fresh env

tests/test_utils_console.py

@@ -1,5 +1,9 @@
+import sys
+from types import ModuleType
+
 import pytest
 
+from scrapy.utils import console
 from scrapy.utils.console import get_shell_embed_func
 
 try:
@@ -39,3 +43,37 @@ def test_get_shell_embed_func_ipython():
     # default shell should be 'ipython'
     shell = get_shell_embed_func()
     assert shell.__name__ == "_embed_ipython_shell"
+
+
+def test_embed_ipython_shell_applies_nest_asyncio_with_asyncio_reactor(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    applied: list[bool] = []
+
+    class FakeInteractiveShellEmbed:
+        @classmethod
+        def clear_instance(cls) -> None:
+            pass
+
+        @classmethod
+        def instance(cls, **kwargs):
+            return lambda: None
+
+    embed_module = ModuleType("IPython.terminal.embed")
+    embed_module.InteractiveShellEmbed = FakeInteractiveShellEmbed
+    ipapp_module = ModuleType("IPython.terminal.ipapp")
+    ipapp_module.load_default_config = object
+    nest_asyncio_module = ModuleType("nest_asyncio")
+    nest_asyncio_module.apply = lambda: applied.append(True)
+
+    monkeypatch.setitem(sys.modules, "IPython.terminal.embed", embed_module)
+    monkeypatch.setitem(sys.modules, "IPython.terminal.ipapp", ipapp_module)
+    monkeypatch.setitem(sys.modules, "nest_asyncio", nest_asyncio_module)
+    monkeypatch.setattr(
+        console, "is_asyncio_reactor_installed", lambda: True, raising=False
+    )
+
+    shell = console._embed_ipython_shell()
+    shell(namespace={}, banner="")
+
+    assert applied == [True]