Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/_core/_eventloop.py: 37%

1from __future__ import annotations 

2 

3import math 

4import sys 

5import threading 

6from collections.abc import Awaitable, Callable, Generator 

7from contextlib import contextmanager 

8from contextvars import Token 

9from importlib import import_module 

10from typing import TYPE_CHECKING, Any, TypeVar 

11 

12from ._exceptions import NoEventLoopError 

13 

14if sys.version_info >= (3, 11): 

15 from typing import TypeVarTuple, Unpack 

16else: 

17 from typing_extensions import TypeVarTuple, Unpack 

18 

19sniffio: Any 

20try: 

21 import sniffio 

22except ModuleNotFoundError: 

23 sniffio = None 

24 

25if TYPE_CHECKING: 

26 from ..abc import AsyncBackend 

27 

28# This must be updated when new backends are introduced 

29BACKENDS = "asyncio", "trio" 

30 

31T_Retval = TypeVar("T_Retval") 

32PosArgsT = TypeVarTuple("PosArgsT") 

33 

34threadlocals = threading.local() 

35loaded_backends: dict[str, type[AsyncBackend]] = {} 

36 

37 

38def run( 

39 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]], 

40 *args: Unpack[PosArgsT], 

41 backend: str = "asyncio", 

42 backend_options: dict[str, Any] | None = None, 

43) -> T_Retval: 

44 """ 

45 Run the given coroutine function in an asynchronous event loop. 

46 

47 The current thread must not be already running an event loop. 

48 

49 :param func: a coroutine function 

50 :param args: positional arguments to ``func`` 

51 :param backend: name of the asynchronous event loop implementation – currently 

52 either ``asyncio`` or ``trio`` 

53 :param backend_options: keyword arguments to call the backend ``run()`` 

54 implementation with (documented :ref:`here <backend options>`) 

55 :return: the return value of the coroutine function 

56 :raises RuntimeError: if an asynchronous event loop is already running in this 

57 thread 

58 :raises LookupError: if the named backend is not found 

59 

60 """ 

61 if asynclib_name := current_async_library(): 

62 raise RuntimeError(f"Already running {asynclib_name} in this thread") 

63 

64 try: 

65 async_backend = get_async_backend(backend) 

66 except ImportError as exc: 

67 if backend in BACKENDS: 

68 raise LookupError( 

69 f"Backend {backend!r} is not available. " 

70 f"Install it with: pip install anyio[{backend}]" 

71 ) from exc 

72 

73 raise LookupError(f"No such backend: {backend}") from exc 

74 

75 token = None 

76 if asynclib_name is None: 

77 # Since we're in control of the event loop, we can cache the name of the async 

78 # library 

79 token = set_current_async_library(backend) 

80 

81 try: 

82 backend_options = backend_options or {} 

83 return async_backend.run(func, args, {}, backend_options) 

84 finally: 

85 reset_current_async_library(token) 

86 

87 

88async def sleep(delay: float) -> None: 

89 """ 

90 Pause the current task for the specified duration. 

91 

92 :param delay: the duration, in seconds 

93 

94 """ 

95 return await get_async_backend().sleep(delay) 

96 

97 

98async def sleep_forever() -> None: 

99 """ 

100 Pause the current task until it's cancelled. 

101 

102 This is a shortcut for ``sleep(math.inf)``. 

103 

104 .. versionadded:: 3.1 

105 

106 """ 

107 await sleep(math.inf) 

108 

109 

110async def sleep_until(deadline: float) -> None: 

111 """ 

112 Pause the current task until the given time. 

113 

114 :param deadline: the absolute time to wake up at (according to the internal 

115 monotonic clock of the event loop) 

116 

117 .. versionadded:: 3.1 

118 

119 """ 

120 now = current_time() 

121 await sleep(max(deadline - now, 0)) 

122 

123 

124def current_time() -> float: 

125 """ 

126 Return the current value of the event loop's internal clock. 

127 

128 :return: the clock value (seconds) 

129 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

130 current thread 

131 

132 """ 

133 return get_async_backend().current_time() 

134 

135 

136def get_all_backends() -> tuple[str, ...]: 

137 """Return a tuple of the names of all built-in backends.""" 

138 return BACKENDS 

139 

140 

141def get_available_backends() -> tuple[str, ...]: 

142 """ 

143 Test for the availability of built-in backends. 

144 

145 :return a tuple of the built-in backend names that were successfully imported 

146 

147 .. versionadded:: 4.12 

148 

149 """ 

150 available_backends: list[str] = [] 

151 for backend_name in get_all_backends(): 

152 try: 

153 get_async_backend(backend_name) 

154 except ImportError: 

155 continue 

156 

157 available_backends.append(backend_name) 

158 

159 return tuple(available_backends) 

160 

161 

162def get_cancelled_exc_class() -> type[BaseException]: 

163 """ 

164 Return the current async library's cancellation exception class. 

165 

166 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

167 current thread 

168 

169 """ 

170 return get_async_backend().cancelled_exception_class() 

171 

172 

173# 

174# Private API 

175# 

176 

177 

178@contextmanager 

179def claim_worker_thread( 

180 backend_class: type[AsyncBackend], token: object 

181) -> Generator[Any, None, None]: 

182 from ..lowlevel import EventLoopToken 

183 

184 threadlocals.current_token = EventLoopToken(backend_class, token) 

185 try: 

186 yield 

187 finally: 

188 del threadlocals.current_token 

189 

190 

191def get_async_backend(asynclib_name: str | None = None) -> type[AsyncBackend]: 

192 if asynclib_name is None: 

193 asynclib_name = current_async_library() 

194 if not asynclib_name: 

195 raise NoEventLoopError( 

196 f"Not currently running on any asynchronous event loop. " 

197 f"Available async backends: {', '.join(get_all_backends())}" 

198 ) 

199 

200 # We use our own dict instead of sys.modules to get the already imported back-end 

201 # class because the appropriate modules in sys.modules could potentially be only 

202 # partially initialized 

203 try: 

204 return loaded_backends[asynclib_name] 

205 except KeyError: 

206 module = import_module(f"anyio._backends._{asynclib_name}") 

207 loaded_backends[asynclib_name] = module.backend_class 

208 return module.backend_class 

209 

210 

211def current_async_library() -> str | None: 

212 if sniffio is None: 

213 # If sniffio is not installed, we assume we're either running asyncio or nothing 

214 import asyncio 

215 

216 try: 

217 asyncio.get_running_loop() 

218 return "asyncio" 

219 except RuntimeError: 

220 pass 

221 else: 

222 try: 

223 return sniffio.current_async_library() 

224 except sniffio.AsyncLibraryNotFoundError: 

225 pass 

226 

227 return None 

228 

229 

230def set_current_async_library(asynclib_name: str | None) -> Token | None: 

231 # no-op if sniffio is not installed 

232 if sniffio is None: 

233 return None 

234 

235 return sniffio.current_async_library_cvar.set(asynclib_name) 

236 

237 

238def reset_current_async_library(token: Token | None) -> None: 

239 if token is not None: 

240 sniffio.current_async_library_cvar.reset(token)