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 

13import ast 

14import asyncio 

15import inspect 

16from functools import wraps 

17 

18_asyncio_event_loop = None 

19 

20 

21def get_asyncio_loop(): 

22 """asyncio has deprecated get_event_loop 

23 

24 Replicate it here, with our desired semantics: 

25 

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 

31 

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 

40 

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 

48 

49 

50class _AsyncIORunner: 

51 def __call__(self, coro): 

52 """ 

53 Handler for asyncio autoawait 

54 """ 

55 return get_asyncio_loop().run_until_complete(coro) 

56 

57 def __str__(self): 

58 return "asyncio" 

59 

60 

61_asyncio_runner = _AsyncIORunner() 

62 

63 

64class _AsyncIOProxy: 

65 """Proxy-object for an asyncio 

66 

67 Any coroutine methods will be wrapped in event_loop.run_ 

68 """ 

69 

70 def __init__(self, obj, event_loop): 

71 self._obj = obj 

72 self._event_loop = event_loop 

73 

74 def __repr__(self): 

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

76 

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) 

88 

89 return _wrapped 

90 else: 

91 return attr 

92 

93 def __dir__(self): 

94 return dir(self._obj) 

95 

96 

97def _curio_runner(coroutine): 

98 """ 

99 handler for curio autoawait 

100 """ 

101 import curio 

102 

103 return curio.run(coroutine) 

104 

105 

106def _trio_runner(async_fn): 

107 import trio 

108 

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 

115 

116 return trio.run(loc, async_fn) 

117 

118 

119def _pseudo_sync_runner(coro): 

120 """ 

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

122 

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

124 

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 ) 

136 

137 

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

139 """Detect if a block of code need to be wrapped in an `async def` 

140 

141 Attempt to parse the block of code, it it compile we're fine. 

142 Otherwise we wrap if and try to compile. 

143 

144 If it works, assume it should be async. Otherwise Return False. 

145 

146 Not handled yet: If the block of code has a return statement as the top 

147 level, it will be seen as async. This is a know limitation. 

148 """ 

149 try: 

150 code = compile( 

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

152 ) 

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

154 except (SyntaxError, MemoryError): 

155 return False