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

1from __future__ import annotations 

2 

3import math 

4import sys 

5from abc import ABCMeta, abstractmethod 

6from collections.abc import ( 

7 AsyncIterator, 

8 Awaitable, 

9 Callable, 

10 Coroutine, 

11 Iterable, 

12 Mapping, 

13 Sequence, 

14) 

15from contextlib import AbstractContextManager 

16from os import PathLike 

17from signal import Signals 

18from socket import AddressFamily, SocketKind, socket 

19from typing import ( 

20 IO, 

21 TYPE_CHECKING, 

22 Any, 

23 Literal, 

24 TypeAlias, 

25 TypeVar, 

26 overload, 

27) 

28 

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

30 from typing import TypeVarTuple, Unpack 

31else: 

32 from typing_extensions import TypeVarTuple, Unpack 

33 

34if TYPE_CHECKING: 

35 from _typeshed import FileDescriptorLike 

36 

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

38 from .._core._tasks import CancelScope 

39 from .._core._testing import TaskInfo 

40 from ._sockets import ( 

41 ConnectedUDPSocket, 

42 ConnectedUNIXDatagramSocket, 

43 IPSockAddrType, 

44 SocketListener, 

45 SocketStream, 

46 UDPSocket, 

47 UNIXDatagramSocket, 

48 UNIXSocketStream, 

49 ) 

50 from ._subprocesses import Process 

51 from ._tasks import TaskGroup 

52 from ._testing import TestRunner 

53 

54T_Retval = TypeVar("T_Retval") 

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

56PosArgsT = TypeVarTuple("PosArgsT") 

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

58 

59 

60class AsyncBackend(metaclass=ABCMeta): 

61 @classmethod 

62 @abstractmethod 

63 def run( 

64 cls, 

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

66 args: tuple[Unpack[PosArgsT]], 

67 kwargs: dict[str, Any], 

68 options: dict[str, Any], 

69 ) -> T_Retval: 

70 """ 

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

72 

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

74 

75 :param func: a coroutine function 

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

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

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

79 with 

80 :return: the return value of the coroutine function 

81 """ 

82 

83 @classmethod 

84 @abstractmethod 

85 def current_token(cls) -> object: 

86 """ 

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

88 

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

90 thread 

91 """ 

92 

93 @classmethod 

94 @abstractmethod 

95 def current_time(cls) -> float: 

96 """ 

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

98 

99 :return: the clock value (seconds) 

100 """ 

101 

102 @classmethod 

103 @abstractmethod 

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

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

106 

107 @classmethod 

108 @abstractmethod 

109 async def checkpoint(cls) -> None: 

110 """ 

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

112 

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

114 :meth:`cancel_shielded_checkpoint`. 

115 """ 

116 

117 @classmethod 

118 async def checkpoint_if_cancelled(cls) -> None: 

119 """ 

120 Check if the current task group has been cancelled. 

121 

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

123 to be scheduled if not. 

124 

125 """ 

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

127 await cls.checkpoint() 

128 

129 @classmethod 

130 async def cancel_shielded_checkpoint(cls) -> None: 

131 """ 

132 Allow the rescheduling of other tasks. 

133 

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

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

136 

137 """ 

138 with cls.create_cancel_scope(shield=True): 

139 await cls.sleep(0) 

140 

141 @classmethod 

142 @abstractmethod 

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

144 """ 

145 Pause the current task for the specified duration. 

146 

147 :param delay: the duration, in seconds 

148 """ 

149 

150 @classmethod 

151 @abstractmethod 

152 def create_cancel_scope( 

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

154 ) -> CancelScope: 

155 pass 

156 

157 @classmethod 

158 @abstractmethod 

159 def current_effective_deadline(cls) -> float: 

160 """ 

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

162 current task. 

163 

164 :return: 

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

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

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

168 :rtype: float 

169 """ 

170 

171 @classmethod 

172 @abstractmethod 

173 def create_task_group(cls) -> TaskGroup: 

174 pass 

175 

176 @classmethod 

177 @abstractmethod 

178 def create_event(cls) -> Event: 

179 pass 

180 

181 @classmethod 

182 @abstractmethod 

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

184 pass 

185 

186 @classmethod 

187 @abstractmethod 

188 def create_semaphore( 

189 cls, 

190 initial_value: int, 

191 *, 

192 max_value: int | None = None, 

193 fast_acquire: bool = False, 

194 ) -> Semaphore: 

195 pass 

196 

197 @classmethod 

198 @abstractmethod 

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

200 pass 

201 

202 @classmethod 

203 @abstractmethod 

204 async def run_sync_in_worker_thread( 

205 cls, 

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

207 args: tuple[Unpack[PosArgsT]], 

208 abandon_on_cancel: bool = False, 

209 limiter: CapacityLimiter | None = None, 

210 ) -> T_Retval: 

211 pass 

212 

213 @classmethod 

214 @abstractmethod 

215 def check_cancelled(cls) -> None: 

216 pass 

217 

218 @classmethod 

219 @abstractmethod 

220 def run_async_from_thread( 

221 cls, 

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

223 args: tuple[Unpack[PosArgsT]], 

224 token: object, 

225 ) -> T_co: 

226 pass 

227 

228 @classmethod 

229 @abstractmethod 

230 def run_sync_from_thread( 

231 cls, 

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

233 args: tuple[Unpack[PosArgsT]], 

234 token: object, 

235 ) -> T_Retval: 

236 pass 

237 

238 @classmethod 

239 @abstractmethod 

240 def run_sync_soon( 

241 cls, 

242 func: Callable[[Unpack[PosArgsT]], object], 

243 args: tuple[Unpack[PosArgsT]], 

244 token: object, 

245 ) -> None: 

246 """Schedule a synchronous callable in an event loop without waiting.""" 

247 

248 @classmethod 

249 @abstractmethod 

250 async def open_process( 

251 cls, 

252 command: StrOrBytesPath | Sequence[StrOrBytesPath], 

253 *, 

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

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

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

257 cwd: StrOrBytesPath | None = None, 

258 env: Mapping[str, str] | None = None, 

259 startupinfo: Any = None, 

260 creationflags: int = 0, 

261 start_new_session: bool = False, 

262 pass_fds: Sequence[int] = (), 

263 user: str | int | None = None, 

264 group: str | int | None = None, 

265 extra_groups: Iterable[str | int] | None = None, 

266 umask: int = -1, 

267 **kwargs: Any, 

268 ) -> Process: 

269 pass 

270 

271 @classmethod 

272 @abstractmethod 

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

274 pass 

275 

276 @classmethod 

277 @abstractmethod 

278 async def connect_tcp( 

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

280 ) -> SocketStream: 

281 pass 

282 

283 @classmethod 

284 @abstractmethod 

285 async def connect_unix( 

286 cls, 

287 path: str | bytes, 

288 kind: Literal[SocketKind.SOCK_STREAM, SocketKind.SOCK_SEQPACKET], 

289 ) -> UNIXSocketStream: 

290 pass 

291 

292 @classmethod 

293 @abstractmethod 

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

295 pass 

296 

297 @classmethod 

298 @abstractmethod 

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

300 pass 

301 

302 @classmethod 

303 @abstractmethod 

304 async def create_udp_socket( 

305 cls, 

306 family: AddressFamily, 

307 local_address: IPSockAddrType | None, 

308 remote_address: IPSockAddrType | None, 

309 reuse_port: bool, 

310 ) -> UDPSocket | ConnectedUDPSocket: 

311 pass 

312 

313 @classmethod 

314 @overload 

315 async def create_unix_datagram_socket( 

316 cls, raw_socket: socket, remote_path: None 

317 ) -> UNIXDatagramSocket: ... 

318 

319 @classmethod 

320 @overload 

321 async def create_unix_datagram_socket( 

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

323 ) -> ConnectedUNIXDatagramSocket: ... 

324 

325 @classmethod 

326 @abstractmethod 

327 async def create_unix_datagram_socket( 

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

329 ) -> UNIXDatagramSocket | ConnectedUNIXDatagramSocket: 

330 pass 

331 

332 @classmethod 

333 @abstractmethod 

334 async def getaddrinfo( 

335 cls, 

336 host: bytes | str | None, 

337 port: str | int | None, 

338 *, 

339 family: int | AddressFamily = 0, 

340 type: int | SocketKind = 0, 

341 proto: int = 0, 

342 flags: int = 0, 

343 ) -> Sequence[ 

344 tuple[ 

345 AddressFamily, 

346 SocketKind, 

347 int, 

348 str, 

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

350 ] 

351 ]: 

352 pass 

353 

354 @classmethod 

355 @abstractmethod 

356 async def getnameinfo( 

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

358 ) -> tuple[str, str]: 

359 pass 

360 

361 @classmethod 

362 @abstractmethod 

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

364 pass 

365 

366 @classmethod 

367 @abstractmethod 

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

369 pass 

370 

371 @classmethod 

372 @abstractmethod 

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

374 pass 

375 

376 @classmethod 

377 @abstractmethod 

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

379 pass 

380 

381 @classmethod 

382 @abstractmethod 

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

384 pass 

385 

386 @classmethod 

387 @abstractmethod 

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

389 pass 

390 

391 @classmethod 

392 @abstractmethod 

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

394 pass 

395 

396 @classmethod 

397 @abstractmethod 

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

399 pass 

400 

401 @classmethod 

402 @abstractmethod 

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

404 pass 

405 

406 @classmethod 

407 @abstractmethod 

408 async def wrap_connected_unix_datagram_socket( 

409 cls, sock: socket 

410 ) -> ConnectedUNIXDatagramSocket: 

411 pass 

412 

413 @classmethod 

414 @abstractmethod 

415 def current_default_thread_limiter(cls) -> CapacityLimiter: 

416 pass 

417 

418 @classmethod 

419 @abstractmethod 

420 def open_signal_receiver( 

421 cls, *signals: Signals 

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

423 pass 

424 

425 @classmethod 

426 @abstractmethod 

427 def get_current_task(cls) -> TaskInfo: 

428 pass 

429 

430 @classmethod 

431 @abstractmethod 

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

433 pass 

434 

435 @classmethod 

436 @abstractmethod 

437 async def wait_all_tasks_blocked(cls) -> None: 

438 pass 

439 

440 @classmethod 

441 @abstractmethod 

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

443 pass