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

1from __future__ import annotations 

2 

3import math 

4import sys 

5from abc import ABCMeta, abstractmethod 

6from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine, Sequence 

7from contextlib import AbstractContextManager 

8from os import PathLike 

9from signal import Signals 

10from socket import AddressFamily, SocketKind, socket 

11from typing import ( 

12 IO, 

13 TYPE_CHECKING, 

14 Any, 

15 TypeAlias, 

16 TypeVar, 

17 overload, 

18) 

19 

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

21 from typing import TypeVarTuple, Unpack 

22else: 

23 from typing_extensions import TypeVarTuple, Unpack 

24 

25if TYPE_CHECKING: 

26 from _typeshed import FileDescriptorLike 

27 

28 from .._core._synchronization import CapacityLimiter, Event, Lock, Semaphore 

29 from .._core._tasks import CancelScope 

30 from .._core._testing import TaskInfo 

31 from ._sockets import ( 

32 ConnectedUDPSocket, 

33 ConnectedUNIXDatagramSocket, 

34 IPSockAddrType, 

35 SocketListener, 

36 SocketStream, 

37 UDPSocket, 

38 UNIXDatagramSocket, 

39 UNIXSocketStream, 

40 ) 

41 from ._subprocesses import Process 

42 from ._tasks import TaskGroup 

43 from ._testing import TestRunner 

44 

45T_Retval = TypeVar("T_Retval") 

46T_co = TypeVar("T_co", covariant=True) 

47PosArgsT = TypeVarTuple("PosArgsT") 

48StrOrBytesPath: TypeAlias = str | bytes | PathLike[str] | PathLike[bytes] 

49 

50 

51class AsyncBackend(metaclass=ABCMeta): 

52 @classmethod 

53 @abstractmethod 

54 def run( 

55 cls, 

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

57 args: tuple[Unpack[PosArgsT]], 

58 kwargs: dict[str, Any], 

59 options: dict[str, Any], 

60 ) -> T_Retval: 

61 """ 

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

63 

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

65 

66 :param func: a coroutine function 

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

68 :param kwargs: positional arguments to ``func`` 

69 :param options: keyword arguments to call the backend ``run()`` implementation 

70 with 

71 :return: the return value of the coroutine function 

72 """ 

73 

74 @classmethod 

75 @abstractmethod 

76 def current_token(cls) -> object: 

77 """ 

78 Return an object that allows other threads to run code inside the event loop. 

79 

80 :return: a token object, specific to the event loop running in the current 

81 thread 

82 """ 

83 

84 @classmethod 

85 @abstractmethod 

86 def current_time(cls) -> float: 

87 """ 

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

89 

90 :return: the clock value (seconds) 

91 """ 

92 

93 @classmethod 

94 @abstractmethod 

95 def cancelled_exception_class(cls) -> type[BaseException]: 

96 """Return the exception class that is raised in a task if it's cancelled.""" 

97 

98 @classmethod 

99 @abstractmethod 

100 async def checkpoint(cls) -> None: 

101 """ 

102 Check if the task has been cancelled, and allow rescheduling of other tasks. 

103 

104 This is effectively the same as running :meth:`checkpoint_if_cancelled` and then 

105 :meth:`cancel_shielded_checkpoint`. 

106 """ 

107 

108 @classmethod 

109 async def checkpoint_if_cancelled(cls) -> None: 

110 """ 

111 Check if the current task group has been cancelled. 

112 

113 This will check if the task has been cancelled, but will not allow other tasks 

114 to be scheduled if not. 

115 

116 """ 

117 if cls.current_effective_deadline() == -math.inf: 

118 await cls.checkpoint() 

119 

120 @classmethod 

121 async def cancel_shielded_checkpoint(cls) -> None: 

122 """ 

123 Allow the rescheduling of other tasks. 

124 

125 This will give other tasks the opportunity to run, but without checking if the 

126 current task group has been cancelled, unlike with :meth:`checkpoint`. 

127 

128 """ 

129 with cls.create_cancel_scope(shield=True): 

130 await cls.sleep(0) 

131 

132 @classmethod 

133 @abstractmethod 

134 async def sleep(cls, delay: float) -> None: 

135 """ 

136 Pause the current task for the specified duration. 

137 

138 :param delay: the duration, in seconds 

139 """ 

140 

141 @classmethod 

142 @abstractmethod 

143 def create_cancel_scope( 

144 cls, *, deadline: float = math.inf, shield: bool = False 

145 ) -> CancelScope: 

146 pass 

147 

148 @classmethod 

149 @abstractmethod 

150 def current_effective_deadline(cls) -> float: 

151 """ 

152 Return the nearest deadline among all the cancel scopes effective for the 

153 current task. 

154 

155 :return: 

156 - a clock value from the event loop's internal clock 

157 - ``inf`` if there is no deadline in effect 

158 - ``-inf`` if the current scope has been cancelled 

159 :rtype: float 

160 """ 

161 

162 @classmethod 

163 @abstractmethod 

164 def create_task_group(cls) -> TaskGroup: 

165 pass 

166 

167 @classmethod 

168 @abstractmethod 

169 def create_event(cls) -> Event: 

170 pass 

171 

172 @classmethod 

173 @abstractmethod 

174 def create_lock(cls, *, fast_acquire: bool) -> Lock: 

175 pass 

176 

177 @classmethod 

178 @abstractmethod 

179 def create_semaphore( 

180 cls, 

181 initial_value: int, 

182 *, 

183 max_value: int | None = None, 

184 fast_acquire: bool = False, 

185 ) -> Semaphore: 

186 pass 

187 

188 @classmethod 

189 @abstractmethod 

190 def create_capacity_limiter(cls, total_tokens: float) -> CapacityLimiter: 

191 pass 

192 

193 @classmethod 

194 @abstractmethod 

195 async def run_sync_in_worker_thread( 

196 cls, 

197 func: Callable[[Unpack[PosArgsT]], T_Retval], 

198 args: tuple[Unpack[PosArgsT]], 

199 abandon_on_cancel: bool = False, 

200 limiter: CapacityLimiter | None = None, 

201 ) -> T_Retval: 

202 pass 

203 

204 @classmethod 

205 @abstractmethod 

206 def check_cancelled(cls) -> None: 

207 pass 

208 

209 @classmethod 

210 @abstractmethod 

211 def run_async_from_thread( 

212 cls, 

213 func: Callable[[Unpack[PosArgsT]], Coroutine[Any, Any, T_co]], 

214 args: tuple[Unpack[PosArgsT]], 

215 token: object, 

216 ) -> T_co: 

217 pass 

218 

219 @classmethod 

220 @abstractmethod 

221 def run_sync_from_thread( 

222 cls, 

223 func: Callable[[Unpack[PosArgsT]], T_Retval], 

224 args: tuple[Unpack[PosArgsT]], 

225 token: object, 

226 ) -> T_Retval: 

227 pass 

228 

229 @classmethod 

230 @abstractmethod 

231 async def open_process( 

232 cls, 

233 command: StrOrBytesPath | Sequence[StrOrBytesPath], 

234 *, 

235 stdin: int | IO[Any] | None, 

236 stdout: int | IO[Any] | None, 

237 stderr: int | IO[Any] | None, 

238 **kwargs: Any, 

239 ) -> Process: 

240 pass 

241 

242 @classmethod 

243 @abstractmethod 

244 def setup_process_pool_exit_at_shutdown(cls, workers: set[Process]) -> None: 

245 pass 

246 

247 @classmethod 

248 @abstractmethod 

249 async def connect_tcp( 

250 cls, host: str, port: int, local_address: IPSockAddrType | None = None 

251 ) -> SocketStream: 

252 pass 

253 

254 @classmethod 

255 @abstractmethod 

256 async def connect_unix(cls, path: str | bytes) -> UNIXSocketStream: 

257 pass 

258 

259 @classmethod 

260 @abstractmethod 

261 def create_tcp_listener(cls, sock: socket) -> SocketListener: 

262 pass 

263 

264 @classmethod 

265 @abstractmethod 

266 def create_unix_listener(cls, sock: socket) -> SocketListener: 

267 pass 

268 

269 @classmethod 

270 @abstractmethod 

271 async def create_udp_socket( 

272 cls, 

273 family: AddressFamily, 

274 local_address: IPSockAddrType | None, 

275 remote_address: IPSockAddrType | None, 

276 reuse_port: bool, 

277 ) -> UDPSocket | ConnectedUDPSocket: 

278 pass 

279 

280 @classmethod 

281 @overload 

282 async def create_unix_datagram_socket( 

283 cls, raw_socket: socket, remote_path: None 

284 ) -> UNIXDatagramSocket: ... 

285 

286 @classmethod 

287 @overload 

288 async def create_unix_datagram_socket( 

289 cls, raw_socket: socket, remote_path: str | bytes 

290 ) -> ConnectedUNIXDatagramSocket: ... 

291 

292 @classmethod 

293 @abstractmethod 

294 async def create_unix_datagram_socket( 

295 cls, raw_socket: socket, remote_path: str | bytes | None 

296 ) -> UNIXDatagramSocket | ConnectedUNIXDatagramSocket: 

297 pass 

298 

299 @classmethod 

300 @abstractmethod 

301 async def getaddrinfo( 

302 cls, 

303 host: bytes | str | None, 

304 port: str | int | None, 

305 *, 

306 family: int | AddressFamily = 0, 

307 type: int | SocketKind = 0, 

308 proto: int = 0, 

309 flags: int = 0, 

310 ) -> Sequence[ 

311 tuple[ 

312 AddressFamily, 

313 SocketKind, 

314 int, 

315 str, 

316 tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes], 

317 ] 

318 ]: 

319 pass 

320 

321 @classmethod 

322 @abstractmethod 

323 async def getnameinfo( 

324 cls, sockaddr: IPSockAddrType, flags: int = 0 

325 ) -> tuple[str, str]: 

326 pass 

327 

328 @classmethod 

329 @abstractmethod 

330 async def wait_readable(cls, obj: FileDescriptorLike) -> None: 

331 pass 

332 

333 @classmethod 

334 @abstractmethod 

335 async def wait_writable(cls, obj: FileDescriptorLike) -> None: 

336 pass 

337 

338 @classmethod 

339 @abstractmethod 

340 def notify_closing(cls, obj: FileDescriptorLike) -> None: 

341 pass 

342 

343 @classmethod 

344 @abstractmethod 

345 async def wrap_listener_socket(cls, sock: socket) -> SocketListener: 

346 pass 

347 

348 @classmethod 

349 @abstractmethod 

350 async def wrap_stream_socket(cls, sock: socket) -> SocketStream: 

351 pass 

352 

353 @classmethod 

354 @abstractmethod 

355 async def wrap_unix_stream_socket(cls, sock: socket) -> UNIXSocketStream: 

356 pass 

357 

358 @classmethod 

359 @abstractmethod 

360 async def wrap_udp_socket(cls, sock: socket) -> UDPSocket: 

361 pass 

362 

363 @classmethod 

364 @abstractmethod 

365 async def wrap_connected_udp_socket(cls, sock: socket) -> ConnectedUDPSocket: 

366 pass 

367 

368 @classmethod 

369 @abstractmethod 

370 async def wrap_unix_datagram_socket(cls, sock: socket) -> UNIXDatagramSocket: 

371 pass 

372 

373 @classmethod 

374 @abstractmethod 

375 async def wrap_connected_unix_datagram_socket( 

376 cls, sock: socket 

377 ) -> ConnectedUNIXDatagramSocket: 

378 pass 

379 

380 @classmethod 

381 @abstractmethod 

382 def current_default_thread_limiter(cls) -> CapacityLimiter: 

383 pass 

384 

385 @classmethod 

386 @abstractmethod 

387 def open_signal_receiver( 

388 cls, *signals: Signals 

389 ) -> AbstractContextManager[AsyncIterator[Signals]]: 

390 pass 

391 

392 @classmethod 

393 @abstractmethod 

394 def get_current_task(cls) -> TaskInfo: 

395 pass 

396 

397 @classmethod 

398 @abstractmethod 

399 def get_running_tasks(cls) -> Sequence[TaskInfo]: 

400 pass 

401 

402 @classmethod 

403 @abstractmethod 

404 async def wait_all_tasks_blocked(cls) -> None: 

405 pass 

406 

407 @classmethod 

408 @abstractmethod 

409 def create_test_runner(cls, options: dict[str, Any]) -> TestRunner: 

410 pass