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

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 :raises ValueError: if ``delay`` is negative (including ``-inf``) or NaN 

94 

95 """ 

96 if not delay >= 0: 

97 raise ValueError("delay must be a non-negative number") 

98 

99 return await get_async_backend().sleep(delay) 

100 

101 

102async def sleep_forever() -> None: 

103 """ 

104 Pause the current task until it's cancelled. 

105 

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

107 

108 .. versionadded:: 3.1 

109 

110 """ 

111 await sleep(math.inf) 

112 

113 

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

115 """ 

116 Pause the current task until the given time. 

117 

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

119 monotonic clock of the event loop) 

120 

121 .. versionadded:: 3.1 

122 

123 """ 

124 now = current_time() 

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

126 

127 

128def current_time() -> float: 

129 """ 

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

131 

132 :return: the clock value (seconds) 

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

134 current thread 

135 

136 """ 

137 return get_async_backend().current_time() 

138 

139 

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

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

142 return BACKENDS 

143 

144 

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

146 """ 

147 Test for the availability of built-in backends. 

148 

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

150 

151 .. versionadded:: 4.12 

152 

153 """ 

154 available_backends: list[str] = [] 

155 for backend_name in get_all_backends(): 

156 try: 

157 get_async_backend(backend_name) 

158 except ImportError: 

159 continue 

160 

161 available_backends.append(backend_name) 

162 

163 return tuple(available_backends) 

164 

165 

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

167 """ 

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

169 

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

171 current thread 

172 

173 """ 

174 return get_async_backend().cancelled_exception_class() 

175 

176 

177# 

178# Private API 

179# 

180 

181 

182@contextmanager 

183def claim_worker_thread( 

184 backend_class: type[AsyncBackend], token: object 

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

186 from ..lowlevel import EventLoopToken 

187 

188 threadlocals.current_token = EventLoopToken(backend_class, token) 

189 try: 

190 yield 

191 finally: 

192 del threadlocals.current_token 

193 

194 

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

196 if asynclib_name is None: 

197 asynclib_name = current_async_library() 

198 if not asynclib_name: 

199 raise NoEventLoopError( 

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

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

202 ) 

203 

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

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

206 # partially initialized 

207 try: 

208 return loaded_backends[asynclib_name] 

209 except KeyError: 

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

211 loaded_backends[asynclib_name] = module.backend_class 

212 return module.backend_class 

213 

214 

215def current_async_library() -> str | None: 

216 if sniffio is None: 

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

218 import asyncio 

219 

220 try: 

221 asyncio.get_running_loop() 

222 return "asyncio" 

223 except RuntimeError: 

224 pass 

225 else: 

226 try: 

227 return sniffio.current_async_library() 

228 except sniffio.AsyncLibraryNotFoundError: 

229 pass 

230 

231 return None 

232 

233 

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

235 # no-op if sniffio is not installed 

236 if sniffio is None: 

237 return None 

238 

239 return sniffio.current_async_library_cvar.set(asynclib_name) 

240 

241 

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

243 if token is not None: 

244 sniffio.current_async_library_cvar.reset(token)