Coverage for Lib/asyncio/__main__.py: 0%

1import argparse 

2import ast 

3import asyncio 

4import asyncio.tools 

5import concurrent.futures 

6import contextvars 

7import inspect 

8import os 

9import site 

10import sys 

11import threading 

12import types 

13import warnings 

14 

15from _colorize import get_theme 

16from _pyrepl.console import InteractiveColoredConsole 

17 

18from . import futures 

19 

20 

21class AsyncIOInteractiveConsole(InteractiveColoredConsole): 

22 

23 def __init__(self, locals, loop): 

24 super().__init__(locals, filename="<stdin>") 

25 self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT 

26 

27 self.loop = loop 

28 self.context = contextvars.copy_context() 

29 

30 def runcode(self, code): 

31 global return_code 

32 future = concurrent.futures.Future() 

33 

34 def callback(): 

35 global return_code 

36 global repl_future 

37 global keyboard_interrupted 

38 

39 repl_future = None 

40 keyboard_interrupted = False 

41 

42 func = types.FunctionType(code, self.locals) 

43 try: 

44 coro = func() 

45 except SystemExit as se: 

46 return_code = se.code 

47 self.loop.stop() 

48 return 

49 except KeyboardInterrupt as ex: 

50 keyboard_interrupted = True 

51 future.set_exception(ex) 

52 return 

53 except BaseException as ex: 

54 future.set_exception(ex) 

55 return 

56 

57 if not inspect.iscoroutine(coro): 

58 future.set_result(coro) 

59 return 

60 

61 try: 

62 repl_future = self.loop.create_task(coro, context=self.context) 

63 futures._chain_future(repl_future, future) 

64 except BaseException as exc: 

65 future.set_exception(exc) 

66 

67 self.loop.call_soon_threadsafe(callback, context=self.context) 

68 

69 try: 

70 return future.result() 

71 except SystemExit as se: 

72 return_code = se.code 

73 self.loop.stop() 

74 return 

75 except BaseException: 

76 if keyboard_interrupted: 

77 if not CAN_USE_PYREPL: 

78 self.write("\nKeyboardInterrupt\n") 

79 else: 

80 self.showtraceback() 

81 return self.STATEMENT_FAILED 

82 

83class REPLThread(threading.Thread): 

84 

85 def run(self): 

86 global return_code 

87 

88 try: 

89 banner = ( 

90 f'asyncio REPL {sys.version} on {sys.platform}\n' 

91 f'Use "await" directly instead of "asyncio.run()".\n' 

92 f'Type "help", "copyright", "credits" or "license" ' 

93 f'for more information.\n' 

94 ) 

95 

96 console.write(banner) 

97 

98 if startup_path := os.getenv("PYTHONSTARTUP"): 

99 sys.audit("cpython.run_startup", startup_path) 

100 

101 import tokenize 

102 with tokenize.open(startup_path) as f: 

103 startup_code = compile(f.read(), startup_path, "exec") 

104 exec(startup_code, console.locals) 

105 

106 ps1 = getattr(sys, "ps1", ">>> ") 

107 if CAN_USE_PYREPL: 

108 theme = get_theme().syntax 

109 ps1 = f"{theme.prompt}{ps1}{theme.reset}" 

110 import_line = f'{theme.keyword}import{theme.reset} asyncio' 

111 else: 

112 import_line = "import asyncio" 

113 console.write(f"{ps1}{import_line}\n") 

114 

115 if CAN_USE_PYREPL: 

116 from _pyrepl.simple_interact import ( 

117 run_multiline_interactive_console, 

118 ) 

119 try: 

120 sys.ps1 = ps1 

121 run_multiline_interactive_console(console) 

122 except SystemExit: 

123 # expected via the `exit` and `quit` commands 

124 pass 

125 except BaseException: 

126 # unexpected issue 

127 console.showtraceback() 

128 console.write("Internal error, ") 

129 return_code = 1 

130 else: 

131 console.interact(banner="", exitmsg="") 

132 finally: 

133 warnings.filterwarnings( 

134 'ignore', 

135 message=r'^coroutine .* was never awaited$', 

136 category=RuntimeWarning) 

137 

138 loop.call_soon_threadsafe(loop.stop) 

139 

140 def interrupt(self) -> None: 

141 if not CAN_USE_PYREPL: 

142 return 

143 

144 from _pyrepl.simple_interact import _get_reader 

145 r = _get_reader() 

146 if r.threading_hook is not None: 

147 r.threading_hook.add("") # type: ignore 

148 

149 

150if __name__ == '__main__': 

151 parser = argparse.ArgumentParser( 

152 prog="python3 -m asyncio", 

153 description="Interactive asyncio shell and CLI tools", 

154 color=True, 

155 ) 

156 subparsers = parser.add_subparsers(help="sub-commands", dest="command") 

157 ps = subparsers.add_parser( 

158 "ps", help="Display a table of all pending tasks in a process" 

159 ) 

160 ps.add_argument("pid", type=int, help="Process ID to inspect") 

161 pstree = subparsers.add_parser( 

162 "pstree", help="Display a tree of all pending tasks in a process" 

163 ) 

164 pstree.add_argument("pid", type=int, help="Process ID to inspect") 

165 args = parser.parse_args() 

166 match args.command: 

167 case "ps": 

168 asyncio.tools.display_awaited_by_tasks_table(args.pid) 

169 sys.exit(0) 

170 case "pstree": 

171 asyncio.tools.display_awaited_by_tasks_tree(args.pid) 

172 sys.exit(0) 

173 case None: 

174 pass # continue to the interactive shell 

175 case _: 

176 # shouldn't happen as an invalid command-line wouldn't parse 

177 # but let's keep it for the next person adding a command 

178 print(f"error: unhandled command {args.command}", file=sys.stderr) 

179 parser.print_usage(file=sys.stderr) 

180 sys.exit(1) 

181 

182 sys.audit("cpython.run_stdin") 

183 

184 if os.getenv('PYTHON_BASIC_REPL'): 

185 CAN_USE_PYREPL = False 

186 else: 

187 from _pyrepl.main import CAN_USE_PYREPL 

188 

189 return_code = 0 

190 loop = asyncio.new_event_loop() 

191 asyncio.set_event_loop(loop) 

192 

193 repl_locals = {'asyncio': asyncio} 

194 for key in {'__name__', '__package__', 

195 '__loader__', '__spec__', 

196 '__builtins__', '__file__'}: 

197 repl_locals[key] = locals()[key] 

198 

199 console = AsyncIOInteractiveConsole(repl_locals, loop) 

200 

201 repl_future = None 

202 keyboard_interrupted = False 

203 

204 try: 

205 import readline # NoQA 

206 except ImportError: 

207 readline = None 

208 

209 interactive_hook = getattr(sys, "__interactivehook__", None) 

210 

211 if interactive_hook is not None: 

212 sys.audit("cpython.run_interactivehook", interactive_hook) 

213 interactive_hook() 

214 

215 if interactive_hook is site.register_readline: 

216 # Fix the completer function to use the interactive console locals 

217 try: 

218 import rlcompleter 

219 except: 

220 pass 

221 else: 

222 if readline is not None: 

223 completer = rlcompleter.Completer(console.locals) 

224 readline.set_completer(completer.complete) 

225 

226 repl_thread = REPLThread(name="Interactive thread") 

227 repl_thread.daemon = True 

228 repl_thread.start() 

229 

230 while True: 

231 try: 

232 loop.run_forever() 

233 except KeyboardInterrupt: 

234 keyboard_interrupted = True 

235 if repl_future and not repl_future.done(): 

236 repl_future.cancel() 

237 repl_thread.interrupt() 

238 continue 

239 else: 

240 break 

241 

242 console.write('exiting asyncio REPL...\n') 

243 sys.exit(return_code)