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 ( 

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 TypeAlias, 

24 TypeVar, 

25 overload, 

26) 

27 

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

29 from typing import TypeVarTuple, Unpack 

30else: 

31 from typing_extensions import TypeVarTuple, Unpack 

32 

33if TYPE_CHECKING: 

34 from _typeshed import FileDescriptorLike 

35 

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

37 from .._core._tasks import CancelScope 

38 from .._core._testing import TaskInfo 

39 from ._sockets import ( 

40 ConnectedUDPSocket, 

41 ConnectedUNIXDatagramSocket, 

42 IPSockAddrType, 

43 SocketListener, 

44 SocketStream, 

45 UDPSocket, 

46 UNIXDatagramSocket, 

47 UNIXSocketStream, 

48 ) 

49 from ._subprocesses import Process 

50 from ._tasks import TaskGroup 

51 from ._testing import TestRunner 

52 

53T_Retval = TypeVar("T_Retval") 

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

55PosArgsT = TypeVarTuple("PosArgsT") 

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

57 

58 

59class AsyncBackend(metaclass=ABCMeta): 

60 @classmethod 

61 @abstractmethod 

62 def run( 

63 cls, 

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

65 args: tuple[Unpack[PosArgsT]], 

66 kwargs: dict[str, Any], 

67 options: dict[str, Any], 

68 ) -> T_Retval: 

69 """ 

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

71 

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

73 

74 :param func: a coroutine function 

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

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

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

78 with 

79 :return: the return value of the coroutine function 

80 """ 

81 

82 @classmethod 

83 @abstractmethod 

84 def current_token(cls) -> object: 

85 """ 

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

87 

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

89 thread 

90 """ 

91 

92 @classmethod 

93 @abstractmethod 

94 def current_time(cls) -> float: 

95 """ 

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

97 

98 :return: the clock value (seconds) 

99 """ 

100 

101 @classmethod 

102 @abstractmethod 

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

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

105 

106 @classmethod 

107 @abstractmethod 

108 async def checkpoint(cls) -> None: 

109 """ 

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

111 

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

113 :meth:`cancel_shielded_checkpoint`. 

114 """ 

115 

116 @classmethod 

117 async def checkpoint_if_cancelled(cls) -> None: 

118 """ 

119 Check if the current task group has been cancelled. 

120 

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

122 to be scheduled if not. 

123 

124 """ 

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

126 await cls.checkpoint() 

127 

128 @classmethod 

129 async def cancel_shielded_checkpoint(cls) -> None: 

130 """ 

131 Allow the rescheduling of other tasks. 

132 

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

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

135 

136 """ 

137 with cls.create_cancel_scope(shield=True): 

138 await cls.sleep(0) 

139 

140 @classmethod 

141 @abstractmethod 

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

143 """ 

144 Pause the current task for the specified duration. 

145 

146 :param delay: the duration, in seconds 

147 """ 

148 

149 @classmethod 

150 @abstractmethod 

151 def create_cancel_scope( 

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

153 ) -> CancelScope: 

154 pass 

155 

156 @classmethod 

157 @abstractmethod 

158 def current_effective_deadline(cls) -> float: 

159 """ 

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

161 current task. 

162 

163 :return: 

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

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

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

167 :rtype: float 

168 """ 

169 

170 @classmethod 

171 @abstractmethod 

172 def create_task_group(cls) -> TaskGroup: 

173 pass 

174 

175 @classmethod 

176 @abstractmethod 

177 def create_event(cls) -> Event: 

178 pass 

179 

180 @classmethod 

181 @abstractmethod 

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

183 pass 

184 

185 @classmethod 

186 @abstractmethod 

187 def create_semaphore( 

188 cls, 

189 initial_value: int, 

190 *, 

191 max_value: int | None = None, 

192 fast_acquire: bool = False, 

193 ) -> Semaphore: 

194 pass 

195 

196 @classmethod 

197 @abstractmethod 

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

199 pass 

200 

201 @classmethod 

202 @abstractmethod 

203 async def run_sync_in_worker_thread( 

204 cls, 

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

206 args: tuple[Unpack[PosArgsT]], 

207 abandon_on_cancel: bool = False, 

208 limiter: CapacityLimiter | None = None, 

209 ) -> T_Retval: 

210 pass 

211 

212 @classmethod 

213 @abstractmethod 

214 def check_cancelled(cls) -> None: 

215 pass 

216 

217 @classmethod 

218 @abstractmethod 

219 def run_async_from_thread( 

220 cls, 

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

222 args: tuple[Unpack[PosArgsT]], 

223 token: object, 

224 ) -> T_co: 

225 pass 

226 

227 @classmethod 

228 @abstractmethod 

229 def run_sync_from_thread( 

230 cls, 

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

232 args: tuple[Unpack[PosArgsT]], 

233 token: object, 

234 ) -> T_Retval: 

235 pass 

236 

237 @classmethod 

238 @abstractmethod 

239 async def open_process( 

240 cls, 

241 command: StrOrBytesPath | Sequence[StrOrBytesPath], 

242 *, 

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

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

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

246 cwd: StrOrBytesPath | None = None, 

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

248 startupinfo: Any = None, 

249 creationflags: int = 0, 

250 start_new_session: bool = False, 

251 pass_fds: Sequence[int] = (), 

252 user: str | int | None = None, 

253 group: str | int | None = None, 

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

255 umask: int = -1, 

256 **kwargs: Any, 

257 ) -> Process: 

258 pass 

259 

260 @classmethod 

261 @abstractmethod 

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

263 pass 

264 

265 @classmethod 

266 @abstractmethod 

267 async def connect_tcp( 

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

269 ) -> SocketStream: 

270 pass 

271 

272 @classmethod 

273 @abstractmethod 

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

275 pass 

276 

277 @classmethod 

278 @abstractmethod 

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

280 pass 

281 

282 @classmethod 

283 @abstractmethod 

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

285 pass 

286 

287 @classmethod 

288 @abstractmethod 

289 async def create_udp_socket( 

290 cls, 

291 family: AddressFamily, 

292 local_address: IPSockAddrType | None, 

293 remote_address: IPSockAddrType | None, 

294 reuse_port: bool, 

295 ) -> UDPSocket | ConnectedUDPSocket: 

296 pass 

297 

298 @classmethod 

299 @overload 

300 async def create_unix_datagram_socket( 

301 cls, raw_socket: socket, remote_path: None 

302 ) -> UNIXDatagramSocket: ... 

303 

304 @classmethod 

305 @overload 

306 async def create_unix_datagram_socket( 

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

308 ) -> ConnectedUNIXDatagramSocket: ... 

309 

310 @classmethod 

311 @abstractmethod 

312 async def create_unix_datagram_socket( 

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

314 ) -> UNIXDatagramSocket | ConnectedUNIXDatagramSocket: 

315 pass 

316 

317 @classmethod 

318 @abstractmethod 

319 async def getaddrinfo( 

320 cls, 

321 host: bytes | str | None, 

322 port: str | int | None, 

323 *, 

324 family: int | AddressFamily = 0, 

325 type: int | SocketKind = 0, 

326 proto: int = 0, 

327 flags: int = 0, 

328 ) -> Sequence[ 

329 tuple[ 

330 AddressFamily, 

331 SocketKind, 

332 int, 

333 str, 

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

335 ] 

336 ]: 

337 pass 

338 

339 @classmethod 

340 @abstractmethod 

341 async def getnameinfo( 

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

343 ) -> tuple[str, str]: 

344 pass 

345 

346 @classmethod 

347 @abstractmethod 

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

349 pass 

350 

351 @classmethod 

352 @abstractmethod 

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

354 pass 

355 

356 @classmethod 

357 @abstractmethod 

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

359 pass 

360 

361 @classmethod 

362 @abstractmethod 

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

364 pass 

365 

366 @classmethod 

367 @abstractmethod 

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

369 pass 

370 

371 @classmethod 

372 @abstractmethod 

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

374 pass 

375 

376 @classmethod 

377 @abstractmethod 

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

379 pass 

380 

381 @classmethod 

382 @abstractmethod 

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

384 pass 

385 

386 @classmethod 

387 @abstractmethod 

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

389 pass 

390 

391 @classmethod 

392 @abstractmethod 

393 async def wrap_connected_unix_datagram_socket( 

394 cls, sock: socket 

395 ) -> ConnectedUNIXDatagramSocket: 

396 pass 

397 

398 @classmethod 

399 @abstractmethod 

400 def current_default_thread_limiter(cls) -> CapacityLimiter: 

401 pass 

402 

403 @classmethod 

404 @abstractmethod 

405 def open_signal_receiver( 

406 cls, *signals: Signals 

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

408 pass 

409 

410 @classmethod 

411 @abstractmethod 

412 def get_current_task(cls) -> TaskInfo: 

413 pass 

414 

415 @classmethod 

416 @abstractmethod 

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

418 pass 

419 

420 @classmethod 

421 @abstractmethod 

422 async def wait_all_tasks_blocked(cls) -> None: 

423 pass 

424 

425 @classmethod 

426 @abstractmethod 

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

428 pass