Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Async helper function that are invalid syntax on Python 3.5 and below.
4This code is best effort, and may have edge cases not behaving as expected. In
5particular it contain a number of heuristics to detect whether code is
6effectively async and need to run in an event loop or not.
8Some constructs (like top-level `return`, or `yield`) are taken care of
9explicitly to actually raise a SyntaxError and stay as close as possible to
10Python semantics.
11"""
13import ast
14import asyncio
15import inspect
16from functools import wraps
18_asyncio_event_loop = None
21def get_asyncio_loop():
22 """asyncio has deprecated get_event_loop
24 Replicate it here, with our desired semantics:
26 - always returns a valid, not-closed loop
27 - not thread-local like asyncio's,
28 because we only want one loop for IPython
29 - if called from inside a coroutine (e.g. in ipykernel),
30 return the running loop
32 .. versionadded:: 8.0
33 """
34 try:
35 return asyncio.get_running_loop()
36 except RuntimeError:
37 # not inside a coroutine,
38 # track our own global
39 pass
41 # not thread-local like asyncio's,
42 # because we only track one event loop to run for IPython itself,
43 # always in the main thread.
44 global _asyncio_event_loop
45 if _asyncio_event_loop is None or _asyncio_event_loop.is_closed():
46 _asyncio_event_loop = asyncio.new_event_loop()
47 return _asyncio_event_loop
50class _AsyncIORunner:
51 def __call__(self, coro):
52 """
53 Handler for asyncio autoawait
54 """
55 return get_asyncio_loop().run_until_complete(coro)
57 def __str__(self):
58 return "asyncio"
61_asyncio_runner = _AsyncIORunner()
64class _AsyncIOProxy:
65 """Proxy-object for an asyncio
67 Any coroutine methods will be wrapped in event_loop.run_
68 """
70 def __init__(self, obj, event_loop):
71 self._obj = obj
72 self._event_loop = event_loop
74 def __repr__(self):
75 return f"<_AsyncIOProxy({self._obj!r})>"
77 def __getattr__(self, key):
78 attr = getattr(self._obj, key)
79 if inspect.iscoroutinefunction(attr):
80 # if it's a coroutine method,
81 # return a threadsafe wrapper onto the _current_ asyncio loop
82 @wraps(attr)
83 def _wrapped(*args, **kwargs):
84 concurrent_future = asyncio.run_coroutine_threadsafe(
85 attr(*args, **kwargs), self._event_loop
86 )
87 return asyncio.wrap_future(concurrent_future)
89 return _wrapped
90 else:
91 return attr
93 def __dir__(self):
94 return dir(self._obj)
97def _curio_runner(coroutine):
98 """
99 handler for curio autoawait
100 """
101 import curio
103 return curio.run(coroutine)
106def _trio_runner(async_fn):
107 import trio
109 async def loc(coro):
110 """
111 We need the dummy no-op async def to protect from
112 trio's internal. See https://github.com/python-trio/trio/issues/89
113 """
114 return await coro
116 return trio.run(loc, async_fn)
119def _pseudo_sync_runner(coro):
120 """
121 A runner that does not really allow async execution, and just advance the coroutine.
123 See discussion in https://github.com/python-trio/trio/issues/608,
125 Credit to Nathaniel Smith
126 """
127 try:
128 coro.send(None)
129 except StopIteration as exc:
130 return exc.value
131 else:
132 # TODO: do not raise but return an execution result with the right info.
133 raise RuntimeError(
134 "{coro_name!r} needs a real async loop".format(coro_name=coro.__name__)
135 )
138def _should_be_async(cell: str) -> bool:
139 """Detect if a block of code needs to be wrapped in an `async def`
141 If the code block has a top-level return statement or is otherwise
142 invalid, `False` will be returned.
143 """
144 try:
145 code = compile(
146 cell, "<>", "exec", flags=getattr(ast, "PyCF_ALLOW_TOP_LEVEL_AWAIT", 0x0)
147 )
148 return inspect.CO_COROUTINE & code.co_flags == inspect.CO_COROUTINE
149 except (SyntaxError, ValueError, MemoryError):
150 return False