python - 在异步代码中调用 IPython.embed()(指定事件循环)

Question

IPython 7.5 文档指出:

Change to Nested Embed

The introduction of the ability to run async code had some effect on the IPython.embed() API. By default, embed will not allow you to run asynchronous code unless an event loop is specified.

但是，文档中似乎没有描述如何指定事件循环。

运行:

import IPython
IPython.embed()

进而

In [1]: %autoawait on

In [2]: %autoawait
IPython autoawait is `on`, and set to use `<function _pseudo_sync_runner at 0x00000000066DEF28>`

In [3]: import asyncio

In [4]: await asyncio.sleep(1)

给出:

---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
~\Envs\[redacted]\lib\site-packages\IPython\core\async_helpers.py in _pseudo_sync_runner(coro)
     71         # TODO: do not raise but return an execution result with the right info.
     72         raise RuntimeError(
---> 73             "{coro_name!r} needs a real async loop".format(coro_name=coro.__name__)
     74         )
     75

RuntimeError: 'run_cell_async' needs a real async loop

另一方面，运行:

import IPython
IPython.embed(using='asyncio')

进而:

In [1]: %autoawait on

In [2]: %autoawait
IPython autoawait is `on`, and set to use `asyncio`

In [3]: import asyncio

In [4]: await asyncio.sleep(1)

给出:

---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
~\Envs\[redacted]\lib\site-packages\IPython\core\async_helpers.py in __call__(self, coro)
     25         import asyncio
     26
---> 27         return asyncio.get_event_loop().run_until_complete(coro)
     28
     29     def __str__(self):

c:\python35-64\Lib\asyncio\base_events.py in run_until_complete(self, future)
    452         future.add_done_callback(_run_until_complete_cb)
    453         try:
--> 454             self.run_forever()
    455         except:
    456             if new_task and future.done() and not future.cancelled():

c:\python35-64\Lib\asyncio\base_events.py in run_forever(self)
    406         self._check_closed()
    407         if self.is_running():
--> 408             raise RuntimeError('This event loop is already running')
    409         if events._get_running_loop() is not None:
    410             raise RuntimeError(

RuntimeError: This event loop is already running
```

Answer 1

from IPython import embed
import nest_asyncio
nest_asyncio.apply()

然后 embed(using='asyncio')可能有点工作。我不知道他们为什么不给我们一个真正的解决方案。