cpython/Lib/asyncio/runners.py at 3.13 · python/cpython · GitHub

Name: cpython/Lib/asyncio/runners.py at 3.13 · python/cpython · GitHub
Rating: 4.9 (5904 reviews)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
__all__= ('Runner', 'run')

importcontextvars
importenum
importfunctools
importthreading
importsignal
from . importcoroutines
from . importevents
from . importexceptions
from . importtasks
from . importconstants

class_State(enum.Enum):
CREATED="created"
INITIALIZED="initialized"
CLOSED="closed"


classRunner:
"""A context manager that controls event loop life cycle.

 The context manager always creates a new event loop,
 allows to run async functions inside it,
 and properly finalizes the loop at the context manager exit.

 If debug is True, the event loop will be run in debug mode.
 If loop_factory is passed, it is used for new event loop creation.

 asyncio.run(main(), debug=True)

 is a shortcut for

 with asyncio.Runner(debug=True) as runner:
 runner.run(main())

 The run() method can be called multiple times within the runner's context.

 This can be useful for interactive console (e.g. IPython),
 unittest runners, console tools, -- everywhere when async code
 is called from existing sync framework and where the preferred single
 asyncio.run() call doesn't work.

 """

# Note: the class is final, it is not intended for inheritance.

def__init__(self, *, debug=None, loop_factory=None):
self._state=_State.CREATED
self._debug=debug
self._loop_factory=loop_factory
self._loop=None
self._context=None
self._interrupt_count=0
self._set_event_loop=False

def__enter__(self):
self._lazy_init()
returnself

def__exit__(self, exc_type, exc_val, exc_tb):
self.close()

defclose(self):
"""Shutdown and close event loop."""
ifself._stateisnot_State.INITIALIZED:
return
try:
loop=self._loop
_cancel_all_tasks(loop)
loop.run_until_complete(loop.shutdown_asyncgens())
loop.run_until_complete(
loop.shutdown_default_executor(constants.THREAD_JOIN_TIMEOUT))
finally:
ifself._set_event_loop:
events.set_event_loop(None)
loop.close()
self._loop=None
self._state=_State.CLOSED

defget_loop(self):
"""Return embedded event loop."""
self._lazy_init()
returnself._loop

defrun(self, coro, *, context=None):
"""Run a coroutine inside the embedded event loop."""
ifnotcoroutines.iscoroutine(coro):
raiseValueError("a coroutine was expected, got {!r}".format(coro))

ifevents._get_running_loop() isnotNone:
# fail fast with short traceback
raiseRuntimeError(
"Runner.run() cannot be called from a running event loop")

self._lazy_init()

ifcontextisNone:
context=self._context
task=self._loop.create_task(coro, context=context)

if (threading.current_thread() isthreading.main_thread()
andsignal.getsignal(signal.SIGINT) issignal.default_int_handler
 ):
sigint_handler=functools.partial(self._on_sigint, main_task=task)
try:
signal.signal(signal.SIGINT, sigint_handler)
exceptValueError:
# `signal.signal` may throw if `threading.main_thread` does
# not support signals (e.g. embedded interpreter with signals
# not registered - see gh-91880)
sigint_handler=None
else:
sigint_handler=None

self._interrupt_count=0
try:
returnself._loop.run_until_complete(task)
exceptexceptions.CancelledError:
ifself._interrupt_count>0:
uncancel=getattr(task, "uncancel", None)
ifuncancelisnotNoneanduncancel() ==0:
raiseKeyboardInterrupt()
raise# CancelledError
finally:
if (sigint_handlerisnotNone
andsignal.getsignal(signal.SIGINT) issigint_handler
 ):
signal.signal(signal.SIGINT, signal.default_int_handler)

def_lazy_init(self):
ifself._stateis_State.CLOSED:
raiseRuntimeError("Runner is closed")
ifself._stateis_State.INITIALIZED:
return
ifself._loop_factoryisNone:
self._loop=events.new_event_loop()
ifnotself._set_event_loop:
# Call set_event_loop only once to avoid calling
# attach_loop multiple times on child watchers
events.set_event_loop(self._loop)
self._set_event_loop=True
else:
self._loop=self._loop_factory()
ifself._debugisnotNone:
self._loop.set_debug(self._debug)
self._context=contextvars.copy_context()
self._state=_State.INITIALIZED

def_on_sigint(self, signum, frame, main_task):
self._interrupt_count+=1
ifself._interrupt_count==1andnotmain_task.done():
main_task.cancel()
# wakeup loop if it is blocked by select() with long timeout
self._loop.call_soon_threadsafe(lambda: None)
return
raiseKeyboardInterrupt()


defrun(main, *, debug=None, loop_factory=None):
"""Execute the coroutine and return the result.

 This function runs the passed coroutine, taking care of
 managing the asyncio event loop, finalizing asynchronous
 generators and closing the default executor.

 This function cannot be called when another asyncio event loop is
 running in the same thread.

 If debug is True, the event loop will be run in debug mode.
 If loop_factory is passed, it is used for new event loop creation.

 This function always creates a new event loop and closes it at the end.
 It should be used as a main entry point for asyncio programs, and should
 ideally only be called once.

 The executor is given a timeout duration of 5 minutes to shutdown.
 If the executor hasn't finished within that duration, a warning is
 emitted and the executor is closed.

 Example:

 async def main():
 await asyncio.sleep(1)
 print('hello')

 asyncio.run(main())
 """
ifevents._get_running_loop() isnotNone:
# fail fast with short traceback
raiseRuntimeError(
"asyncio.run() cannot be called from a running event loop")

withRunner(debug=debug, loop_factory=loop_factory) asrunner:
returnrunner.run(main)


def_cancel_all_tasks(loop):
to_cancel=tasks.all_tasks(loop)
ifnotto_cancel:
return

fortaskinto_cancel:
task.cancel()

loop.run_until_complete(tasks.gather(*to_cancel, return_exceptions=True))

fortaskinto_cancel:
iftask.cancelled():
continue
iftask.exception() isnotNone:
loop.call_exception_handler({
'message': 'unhandled exception during asyncio.run() shutdown',
'exception': task.exception(),
'task': task,
 })