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
|
import asyncio as __asyncio
import typing as _typing
import sys as _sys
import warnings as _warnings
from asyncio.events import BaseDefaultEventLoopPolicy as __BasePolicy
from . import includes as __includes # NOQA
from .loop import Loop as __BaseLoop # NOQA
from ._version import __version__ # NOQA
__all__ = ('new_event_loop', 'install', 'EventLoopPolicy')
_T = _typing.TypeVar("_T")
class Loop(__BaseLoop, __asyncio.AbstractEventLoop): # type: ignore[misc]
pass
def new_event_loop() -> Loop:
"""Return a new event loop."""
return Loop()
def install() -> None:
"""A helper function to install uvloop policy."""
if _sys.version_info[:2] >= (3, 12):
_warnings.warn(
'uvloop.install() is deprecated in favor of uvloop.run() '
'starting with Python 3.12.',
DeprecationWarning,
stacklevel=1,
)
__asyncio.set_event_loop_policy(EventLoopPolicy())
if _typing.TYPE_CHECKING:
def run(
main: _typing.Coroutine[_typing.Any, _typing.Any, _T],
*,
loop_factory: _typing.Optional[
_typing.Callable[[], Loop]
] = new_event_loop,
debug: _typing.Optional[bool]=None,
) -> _T:
"""The preferred way of running a coroutine with uvloop."""
else:
def run(main, *, loop_factory=new_event_loop, debug=None, **run_kwargs):
"""The preferred way of running a coroutine with uvloop."""
async def wrapper():
# If `loop_factory` is provided we want it to return
# either uvloop.Loop or a subtype of it, assuming the user
# is using `uvloop.run()` intentionally.
loop = __asyncio._get_running_loop()
if not isinstance(loop, Loop):
raise TypeError('uvloop.run() uses a non-uvloop event loop')
return await main
vi = _sys.version_info[:2]
if vi <= (3, 10):
# Copied from python/cpython
if __asyncio._get_running_loop() is not None:
raise RuntimeError(
"asyncio.run() cannot be called from a running event loop")
if not __asyncio.iscoroutine(main):
raise ValueError(
"a coroutine was expected, got {!r}".format(main)
)
loop = loop_factory()
try:
__asyncio.set_event_loop(loop)
if debug is not None:
loop.set_debug(debug)
return loop.run_until_complete(wrapper())
finally:
try:
_cancel_all_tasks(loop)
loop.run_until_complete(loop.shutdown_asyncgens())
if hasattr(loop, 'shutdown_default_executor'):
loop.run_until_complete(
loop.shutdown_default_executor()
)
finally:
__asyncio.set_event_loop(None)
loop.close()
elif vi == (3, 11):
if __asyncio._get_running_loop() is not None:
raise RuntimeError(
"asyncio.run() cannot be called from a running event loop")
with __asyncio.Runner(
loop_factory=loop_factory,
debug=debug,
**run_kwargs
) as runner:
return runner.run(wrapper())
else:
assert vi >= (3, 12)
return __asyncio.run(
wrapper(),
loop_factory=loop_factory,
debug=debug,
**run_kwargs
)
def _cancel_all_tasks(loop: __asyncio.AbstractEventLoop) -> None:
# Copied from python/cpython
to_cancel = __asyncio.all_tasks(loop)
if not to_cancel:
return
for task in to_cancel:
task.cancel()
loop.run_until_complete(
__asyncio.gather(*to_cancel, return_exceptions=True)
)
for task in to_cancel:
if task.cancelled():
continue
if task.exception() is not None:
loop.call_exception_handler({
'message': 'unhandled exception during asyncio.run() shutdown',
'exception': task.exception(),
'task': task,
})
class EventLoopPolicy(__BasePolicy):
"""Event loop policy.
The preferred way to make your application use uvloop:
>>> import asyncio
>>> import uvloop
>>> asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
>>> asyncio.get_event_loop()
<uvloop.Loop running=False closed=False debug=False>
"""
def _loop_factory(self) -> Loop:
return new_event_loop()
if _typing.TYPE_CHECKING:
# EventLoopPolicy doesn't implement these, but since they are marked
# as abstract in typeshed, we have to put them in so mypy thinks
# the base methods are overridden. This is the same approach taken
# for the Windows event loop policy classes in typeshed.
def get_child_watcher(self) -> _typing.NoReturn:
...
def set_child_watcher(
self, watcher: _typing.Any
) -> _typing.NoReturn:
...
|
