Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/IPython/core/async_helpers.py: 35%

1""" 

2Async helper function that are invalid syntax on Python 3.5 and below. 

3 

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. 

7 

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""" 

12 

13from __future__ import annotations 

14 

15import ast 

16import inspect 

17from functools import wraps 

18from typing import TYPE_CHECKING 

19 

20if TYPE_CHECKING: 

21 import asyncio 

22 

23_asyncio_event_loop: asyncio.AbstractEventLoop | None = None 

24 

25 

26def get_asyncio_loop(): 

27 """asyncio has deprecated get_event_loop 

28 

29 Replicate it here, with our desired semantics: 

30 

31 - always returns a valid, not-closed loop 

32 - not thread-local like asyncio's, 

33 because we only want one loop for IPython 

34 - if called from inside a coroutine (e.g. in ipykernel), 

35 return the running loop 

36 

37 .. versionadded:: 8.0 

38 """ 

39 # asyncio (and everything it drags in) is only imported the first 

40 # time an event loop is actually needed, rather than on every 

41 # IPython startup. 

42 import asyncio 

43 

44 try: 

45 return asyncio.get_running_loop() 

46 except RuntimeError: 

47 # not inside a coroutine, 

48 # track our own global 

49 pass 

50 

51 # not thread-local like asyncio's, 

52 # because we only track one event loop to run for IPython itself, 

53 # always in the main thread. 

54 global _asyncio_event_loop 

55 if _asyncio_event_loop is None or _asyncio_event_loop.is_closed(): 

56 _asyncio_event_loop = asyncio.new_event_loop() 

57 return _asyncio_event_loop 

58 

59 

60class _AsyncIORunner: 

61 def __call__(self, coro): 

62 """ 

63 Handler for asyncio autoawait 

64 """ 

65 return get_asyncio_loop().run_until_complete(coro) 

66 

67 def __str__(self): 

68 return "asyncio" 

69 

70 

71_asyncio_runner = _AsyncIORunner() 

72 

73 

74class _AsyncIOProxy: 

75 """Proxy-object for an asyncio 

76 

77 Any coroutine methods will be wrapped in event_loop.run_ 

78 """ 

79 

80 def __init__(self, obj, event_loop): 

81 self._obj = obj 

82 self._event_loop = event_loop 

83 

84 def __repr__(self): 

85 return f"<_AsyncIOProxy({self._obj!r})>" 

86 

87 def __getattr__(self, key): 

88 attr = getattr(self._obj, key) 

89 if inspect.iscoroutinefunction(attr): 

90 # if it's a coroutine method, 

91 # return a threadsafe wrapper onto the _current_ asyncio loop 

92 @wraps(attr) 

93 def _wrapped(*args, **kwargs): 

94 import asyncio 

95 

96 concurrent_future = asyncio.run_coroutine_threadsafe( 

97 attr(*args, **kwargs), self._event_loop 

98 ) 

99 return asyncio.wrap_future(concurrent_future) 

100 

101 return _wrapped 

102 else: 

103 return attr 

104 

105 def __dir__(self): 

106 return dir(self._obj) 

107 

108 

109def _curio_runner(coroutine): 

110 """ 

111 handler for curio autoawait 

112 """ 

113 import curio 

114 

115 return curio.run(coroutine) 

116 

117 

118def _trio_runner(async_fn): 

119 import trio 

120 

121 async def loc(coro): 

122 """ 

123 We need the dummy no-op async def to protect from 

124 trio's internal. See https://github.com/python-trio/trio/issues/89 

125 """ 

126 return await coro 

127 

128 return trio.run(loc, async_fn) 

129 

130 

131def _pseudo_sync_runner(coro): 

132 """ 

133 A runner that does not really allow async execution, and just advance the coroutine. 

134 

135 See discussion in https://github.com/python-trio/trio/issues/608, 

136 

137 Credit to Nathaniel Smith 

138 """ 

139 try: 

140 coro.send(None) 

141 except StopIteration as exc: 

142 return exc.value 

143 else: 

144 # TODO: do not raise but return an execution result with the right info. 

145 raise RuntimeError( 

146 f"{coro.__name__!r} needs a real async loop" 

147 ) 

148 

149 

150def _should_be_async(cell: str) -> bool: 

151 """Detect if a block of code needs to be wrapped in an `async def` 

152 

153 If the code block has a top-level return statement or is otherwise 

154 invalid, `False` will be returned. 

155 """ 

156 try: 

157 code = compile( 

158 cell, "<>", "exec", flags=getattr(ast, "PyCF_ALLOW_TOP_LEVEL_AWAIT", 0x0) 

159 ) 

160 return inspect.CO_COROUTINE & code.co_flags == inspect.CO_COROUTINE 

161 except (SyntaxError, ValueError, MemoryError): 

162 return False