Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/anyio/_core/_fileio.py: 47%

1from __future__ import annotations 

2 

3import os 

4import pathlib 

5import sys 

6from collections.abc import Callable, Iterable, Iterator, Sequence 

7from dataclasses import dataclass 

8from functools import partial 

9from os import PathLike 

10from typing import ( 

11 IO, 

12 TYPE_CHECKING, 

13 Any, 

14 AnyStr, 

15 AsyncIterator, 

16 Final, 

17 Generic, 

18 cast, 

19 overload, 

20) 

21 

22from .. import to_thread 

23from ..abc import AsyncResource 

24 

25if TYPE_CHECKING: 

26 from _typeshed import OpenBinaryMode, OpenTextMode, ReadableBuffer, WriteableBuffer 

27else: 

28 ReadableBuffer = OpenBinaryMode = OpenTextMode = WriteableBuffer = object 

29 

30 

31class AsyncFile(AsyncResource, Generic[AnyStr]): 

32 """ 

33 An asynchronous file object. 

34 

35 This class wraps a standard file object and provides async friendly versions of the 

36 following blocking methods (where available on the original file object): 

37 

38 * read 

39 * read1 

40 * readline 

41 * readlines 

42 * readinto 

43 * readinto1 

44 * write 

45 * writelines 

46 * truncate 

47 * seek 

48 * tell 

49 * flush 

50 

51 All other methods are directly passed through. 

52 

53 This class supports the asynchronous context manager protocol which closes the 

54 underlying file at the end of the context block. 

55 

56 This class also supports asynchronous iteration:: 

57 

58 async with await open_file(...) as f: 

59 async for line in f: 

60 print(line) 

61 """ 

62 

63 def __init__(self, fp: IO[AnyStr]) -> None: 

64 self._fp: Any = fp 

65 

66 def __getattr__(self, name: str) -> object: 

67 return getattr(self._fp, name) 

68 

69 @property 

70 def wrapped(self) -> IO[AnyStr]: 

71 """The wrapped file object.""" 

72 return self._fp 

73 

74 async def __aiter__(self) -> AsyncIterator[AnyStr]: 

75 while True: 

76 line = await self.readline() 

77 if line: 

78 yield line 

79 else: 

80 break 

81 

82 async def aclose(self) -> None: 

83 return await to_thread.run_sync(self._fp.close) 

84 

85 async def read(self, size: int = -1) -> AnyStr: 

86 return await to_thread.run_sync(self._fp.read, size) 

87 

88 async def read1(self: AsyncFile[bytes], size: int = -1) -> bytes: 

89 return await to_thread.run_sync(self._fp.read1, size) 

90 

91 async def readline(self) -> AnyStr: 

92 return await to_thread.run_sync(self._fp.readline) 

93 

94 async def readlines(self) -> list[AnyStr]: 

95 return await to_thread.run_sync(self._fp.readlines) 

96 

97 async def readinto(self: AsyncFile[bytes], b: WriteableBuffer) -> bytes: 

98 return await to_thread.run_sync(self._fp.readinto, b) 

99 

100 async def readinto1(self: AsyncFile[bytes], b: WriteableBuffer) -> bytes: 

101 return await to_thread.run_sync(self._fp.readinto1, b) 

102 

103 @overload 

104 async def write(self: AsyncFile[bytes], b: ReadableBuffer) -> int: 

105 ... 

106 

107 @overload 

108 async def write(self: AsyncFile[str], b: str) -> int: 

109 ... 

110 

111 async def write(self, b: ReadableBuffer | str) -> int: 

112 return await to_thread.run_sync(self._fp.write, b) 

113 

114 @overload 

115 async def writelines( 

116 self: AsyncFile[bytes], lines: Iterable[ReadableBuffer] 

117 ) -> None: 

118 ... 

119 

120 @overload 

121 async def writelines(self: AsyncFile[str], lines: Iterable[str]) -> None: 

122 ... 

123 

124 async def writelines(self, lines: Iterable[ReadableBuffer] | Iterable[str]) -> None: 

125 return await to_thread.run_sync(self._fp.writelines, lines) 

126 

127 async def truncate(self, size: int | None = None) -> int: 

128 return await to_thread.run_sync(self._fp.truncate, size) 

129 

130 async def seek(self, offset: int, whence: int | None = os.SEEK_SET) -> int: 

131 return await to_thread.run_sync(self._fp.seek, offset, whence) 

132 

133 async def tell(self) -> int: 

134 return await to_thread.run_sync(self._fp.tell) 

135 

136 async def flush(self) -> None: 

137 return await to_thread.run_sync(self._fp.flush) 

138 

139 

140@overload 

141async def open_file( 

142 file: str | PathLike[str] | int, 

143 mode: OpenBinaryMode, 

144 buffering: int = ..., 

145 encoding: str | None = ..., 

146 errors: str | None = ..., 

147 newline: str | None = ..., 

148 closefd: bool = ..., 

149 opener: Callable[[str, int], int] | None = ..., 

150) -> AsyncFile[bytes]: 

151 ... 

152 

153 

154@overload 

155async def open_file( 

156 file: str | PathLike[str] | int, 

157 mode: OpenTextMode = ..., 

158 buffering: int = ..., 

159 encoding: str | None = ..., 

160 errors: str | None = ..., 

161 newline: str | None = ..., 

162 closefd: bool = ..., 

163 opener: Callable[[str, int], int] | None = ..., 

164) -> AsyncFile[str]: 

165 ... 

166 

167 

168async def open_file( 

169 file: str | PathLike[str] | int, 

170 mode: str = "r", 

171 buffering: int = -1, 

172 encoding: str | None = None, 

173 errors: str | None = None, 

174 newline: str | None = None, 

175 closefd: bool = True, 

176 opener: Callable[[str, int], int] | None = None, 

177) -> AsyncFile[Any]: 

178 """ 

179 Open a file asynchronously. 

180 

181 The arguments are exactly the same as for the builtin :func:`open`. 

182 

183 :return: an asynchronous file object 

184 

185 """ 

186 fp = await to_thread.run_sync( 

187 open, file, mode, buffering, encoding, errors, newline, closefd, opener 

188 ) 

189 return AsyncFile(fp) 

190 

191 

192def wrap_file(file: IO[AnyStr]) -> AsyncFile[AnyStr]: 

193 """ 

194 Wrap an existing file as an asynchronous file. 

195 

196 :param file: an existing file-like object 

197 :return: an asynchronous file object 

198 

199 """ 

200 return AsyncFile(file) 

201 

202 

203@dataclass(eq=False) 

204class _PathIterator(AsyncIterator["Path"]): 

205 iterator: Iterator[PathLike[str]] 

206 

207 async def __anext__(self) -> Path: 

208 nextval = await to_thread.run_sync(next, self.iterator, None, cancellable=True) 

209 if nextval is None: 

210 raise StopAsyncIteration from None 

211 

212 return Path(cast("PathLike[str]", nextval)) 

213 

214 

215class Path: 

216 """ 

217 An asynchronous version of :class:`pathlib.Path`. 

218 

219 This class cannot be substituted for :class:`pathlib.Path` or 

220 :class:`pathlib.PurePath`, but it is compatible with the :class:`os.PathLike` 

221 interface. 

222 

223 It implements the Python 3.10 version of :class:`pathlib.Path` interface, except for 

224 the deprecated :meth:`~pathlib.Path.link_to` method. 

225 

226 Any methods that do disk I/O need to be awaited on. These methods are: 

227 

228 * :meth:`~pathlib.Path.absolute` 

229 * :meth:`~pathlib.Path.chmod` 

230 * :meth:`~pathlib.Path.cwd` 

231 * :meth:`~pathlib.Path.exists` 

232 * :meth:`~pathlib.Path.expanduser` 

233 * :meth:`~pathlib.Path.group` 

234 * :meth:`~pathlib.Path.hardlink_to` 

235 * :meth:`~pathlib.Path.home` 

236 * :meth:`~pathlib.Path.is_block_device` 

237 * :meth:`~pathlib.Path.is_char_device` 

238 * :meth:`~pathlib.Path.is_dir` 

239 * :meth:`~pathlib.Path.is_fifo` 

240 * :meth:`~pathlib.Path.is_file` 

241 * :meth:`~pathlib.Path.is_mount` 

242 * :meth:`~pathlib.Path.lchmod` 

243 * :meth:`~pathlib.Path.lstat` 

244 * :meth:`~pathlib.Path.mkdir` 

245 * :meth:`~pathlib.Path.open` 

246 * :meth:`~pathlib.Path.owner` 

247 * :meth:`~pathlib.Path.read_bytes` 

248 * :meth:`~pathlib.Path.read_text` 

249 * :meth:`~pathlib.Path.readlink` 

250 * :meth:`~pathlib.Path.rename` 

251 * :meth:`~pathlib.Path.replace` 

252 * :meth:`~pathlib.Path.rmdir` 

253 * :meth:`~pathlib.Path.samefile` 

254 * :meth:`~pathlib.Path.stat` 

255 * :meth:`~pathlib.Path.touch` 

256 * :meth:`~pathlib.Path.unlink` 

257 * :meth:`~pathlib.Path.write_bytes` 

258 * :meth:`~pathlib.Path.write_text` 

259 

260 Additionally, the following methods return an async iterator yielding 

261 :class:`~.Path` objects: 

262 

263 * :meth:`~pathlib.Path.glob` 

264 * :meth:`~pathlib.Path.iterdir` 

265 * :meth:`~pathlib.Path.rglob` 

266 """ 

267 

268 __slots__ = "_path", "__weakref__" 

269 

270 __weakref__: Any 

271 

272 def __init__(self, *args: str | PathLike[str]) -> None: 

273 self._path: Final[pathlib.Path] = pathlib.Path(*args) 

274 

275 def __fspath__(self) -> str: 

276 return self._path.__fspath__() 

277 

278 def __str__(self) -> str: 

279 return self._path.__str__() 

280 

281 def __repr__(self) -> str: 

282 return f"{self.__class__.__name__}({self.as_posix()!r})" 

283 

284 def __bytes__(self) -> bytes: 

285 return self._path.__bytes__() 

286 

287 def __hash__(self) -> int: 

288 return self._path.__hash__() 

289 

290 def __eq__(self, other: object) -> bool: 

291 target = other._path if isinstance(other, Path) else other 

292 return self._path.__eq__(target) 

293 

294 def __lt__(self, other: Path) -> bool: 

295 target = other._path if isinstance(other, Path) else other 

296 return self._path.__lt__(target) 

297 

298 def __le__(self, other: Path) -> bool: 

299 target = other._path if isinstance(other, Path) else other 

300 return self._path.__le__(target) 

301 

302 def __gt__(self, other: Path) -> bool: 

303 target = other._path if isinstance(other, Path) else other 

304 return self._path.__gt__(target) 

305 

306 def __ge__(self, other: Path) -> bool: 

307 target = other._path if isinstance(other, Path) else other 

308 return self._path.__ge__(target) 

309 

310 def __truediv__(self, other: Any) -> Path: 

311 return Path(self._path / other) 

312 

313 def __rtruediv__(self, other: Any) -> Path: 

314 return Path(other) / self 

315 

316 @property 

317 def parts(self) -> tuple[str, ...]: 

318 return self._path.parts 

319 

320 @property 

321 def drive(self) -> str: 

322 return self._path.drive 

323 

324 @property 

325 def root(self) -> str: 

326 return self._path.root 

327 

328 @property 

329 def anchor(self) -> str: 

330 return self._path.anchor 

331 

332 @property 

333 def parents(self) -> Sequence[Path]: 

334 return tuple(Path(p) for p in self._path.parents) 

335 

336 @property 

337 def parent(self) -> Path: 

338 return Path(self._path.parent) 

339 

340 @property 

341 def name(self) -> str: 

342 return self._path.name 

343 

344 @property 

345 def suffix(self) -> str: 

346 return self._path.suffix 

347 

348 @property 

349 def suffixes(self) -> list[str]: 

350 return self._path.suffixes 

351 

352 @property 

353 def stem(self) -> str: 

354 return self._path.stem 

355 

356 async def absolute(self) -> Path: 

357 path = await to_thread.run_sync(self._path.absolute) 

358 return Path(path) 

359 

360 def as_posix(self) -> str: 

361 return self._path.as_posix() 

362 

363 def as_uri(self) -> str: 

364 return self._path.as_uri() 

365 

366 def match(self, path_pattern: str) -> bool: 

367 return self._path.match(path_pattern) 

368 

369 def is_relative_to(self, other: str | PathLike[str]) -> bool: 

370 try: 

371 self.relative_to(other) 

372 return True 

373 except ValueError: 

374 return False 

375 

376 async def is_junction(self) -> bool: 

377 return await to_thread.run_sync(self._path.is_junction) 

378 

379 async def chmod(self, mode: int, *, follow_symlinks: bool = True) -> None: 

380 func = partial(os.chmod, follow_symlinks=follow_symlinks) 

381 return await to_thread.run_sync(func, self._path, mode) 

382 

383 @classmethod 

384 async def cwd(cls) -> Path: 

385 path = await to_thread.run_sync(pathlib.Path.cwd) 

386 return cls(path) 

387 

388 async def exists(self) -> bool: 

389 return await to_thread.run_sync(self._path.exists, cancellable=True) 

390 

391 async def expanduser(self) -> Path: 

392 return Path(await to_thread.run_sync(self._path.expanduser, cancellable=True)) 

393 

394 def glob(self, pattern: str) -> AsyncIterator[Path]: 

395 gen = self._path.glob(pattern) 

396 return _PathIterator(gen) 

397 

398 async def group(self) -> str: 

399 return await to_thread.run_sync(self._path.group, cancellable=True) 

400 

401 async def hardlink_to(self, target: str | pathlib.Path | Path) -> None: 

402 if isinstance(target, Path): 

403 target = target._path 

404 

405 await to_thread.run_sync(os.link, target, self) 

406 

407 @classmethod 

408 async def home(cls) -> Path: 

409 home_path = await to_thread.run_sync(pathlib.Path.home) 

410 return cls(home_path) 

411 

412 def is_absolute(self) -> bool: 

413 return self._path.is_absolute() 

414 

415 async def is_block_device(self) -> bool: 

416 return await to_thread.run_sync(self._path.is_block_device, cancellable=True) 

417 

418 async def is_char_device(self) -> bool: 

419 return await to_thread.run_sync(self._path.is_char_device, cancellable=True) 

420 

421 async def is_dir(self) -> bool: 

422 return await to_thread.run_sync(self._path.is_dir, cancellable=True) 

423 

424 async def is_fifo(self) -> bool: 

425 return await to_thread.run_sync(self._path.is_fifo, cancellable=True) 

426 

427 async def is_file(self) -> bool: 

428 return await to_thread.run_sync(self._path.is_file, cancellable=True) 

429 

430 async def is_mount(self) -> bool: 

431 return await to_thread.run_sync(os.path.ismount, self._path, cancellable=True) 

432 

433 def is_reserved(self) -> bool: 

434 return self._path.is_reserved() 

435 

436 async def is_socket(self) -> bool: 

437 return await to_thread.run_sync(self._path.is_socket, cancellable=True) 

438 

439 async def is_symlink(self) -> bool: 

440 return await to_thread.run_sync(self._path.is_symlink, cancellable=True) 

441 

442 def iterdir(self) -> AsyncIterator[Path]: 

443 gen = self._path.iterdir() 

444 return _PathIterator(gen) 

445 

446 def joinpath(self, *args: str | PathLike[str]) -> Path: 

447 return Path(self._path.joinpath(*args)) 

448 

449 async def lchmod(self, mode: int) -> None: 

450 await to_thread.run_sync(self._path.lchmod, mode) 

451 

452 async def lstat(self) -> os.stat_result: 

453 return await to_thread.run_sync(self._path.lstat, cancellable=True) 

454 

455 async def mkdir( 

456 self, mode: int = 0o777, parents: bool = False, exist_ok: bool = False 

457 ) -> None: 

458 await to_thread.run_sync(self._path.mkdir, mode, parents, exist_ok) 

459 

460 @overload 

461 async def open( 

462 self, 

463 mode: OpenBinaryMode, 

464 buffering: int = ..., 

465 encoding: str | None = ..., 

466 errors: str | None = ..., 

467 newline: str | None = ..., 

468 ) -> AsyncFile[bytes]: 

469 ... 

470 

471 @overload 

472 async def open( 

473 self, 

474 mode: OpenTextMode = ..., 

475 buffering: int = ..., 

476 encoding: str | None = ..., 

477 errors: str | None = ..., 

478 newline: str | None = ..., 

479 ) -> AsyncFile[str]: 

480 ... 

481 

482 async def open( 

483 self, 

484 mode: str = "r", 

485 buffering: int = -1, 

486 encoding: str | None = None, 

487 errors: str | None = None, 

488 newline: str | None = None, 

489 ) -> AsyncFile[Any]: 

490 fp = await to_thread.run_sync( 

491 self._path.open, mode, buffering, encoding, errors, newline 

492 ) 

493 return AsyncFile(fp) 

494 

495 async def owner(self) -> str: 

496 return await to_thread.run_sync(self._path.owner, cancellable=True) 

497 

498 async def read_bytes(self) -> bytes: 

499 return await to_thread.run_sync(self._path.read_bytes) 

500 

501 async def read_text( 

502 self, encoding: str | None = None, errors: str | None = None 

503 ) -> str: 

504 return await to_thread.run_sync(self._path.read_text, encoding, errors) 

505 

506 def relative_to(self, *other: str | PathLike[str]) -> Path: 

507 return Path(self._path.relative_to(*other)) 

508 

509 async def readlink(self) -> Path: 

510 target = await to_thread.run_sync(os.readlink, self._path) 

511 return Path(cast(str, target)) 

512 

513 async def rename(self, target: str | pathlib.PurePath | Path) -> Path: 

514 if isinstance(target, Path): 

515 target = target._path 

516 

517 await to_thread.run_sync(self._path.rename, target) 

518 return Path(target) 

519 

520 async def replace(self, target: str | pathlib.PurePath | Path) -> Path: 

521 if isinstance(target, Path): 

522 target = target._path 

523 

524 await to_thread.run_sync(self._path.replace, target) 

525 return Path(target) 

526 

527 async def resolve(self, strict: bool = False) -> Path: 

528 func = partial(self._path.resolve, strict=strict) 

529 return Path(await to_thread.run_sync(func, cancellable=True)) 

530 

531 def rglob(self, pattern: str) -> AsyncIterator[Path]: 

532 gen = self._path.rglob(pattern) 

533 return _PathIterator(gen) 

534 

535 async def rmdir(self) -> None: 

536 await to_thread.run_sync(self._path.rmdir) 

537 

538 async def samefile( 

539 self, other_path: str | bytes | int | pathlib.Path | Path 

540 ) -> bool: 

541 if isinstance(other_path, Path): 

542 other_path = other_path._path 

543 

544 return await to_thread.run_sync( 

545 self._path.samefile, other_path, cancellable=True 

546 ) 

547 

548 async def stat(self, *, follow_symlinks: bool = True) -> os.stat_result: 

549 func = partial(os.stat, follow_symlinks=follow_symlinks) 

550 return await to_thread.run_sync(func, self._path, cancellable=True) 

551 

552 async def symlink_to( 

553 self, 

554 target: str | pathlib.Path | Path, 

555 target_is_directory: bool = False, 

556 ) -> None: 

557 if isinstance(target, Path): 

558 target = target._path 

559 

560 await to_thread.run_sync(self._path.symlink_to, target, target_is_directory) 

561 

562 async def touch(self, mode: int = 0o666, exist_ok: bool = True) -> None: 

563 await to_thread.run_sync(self._path.touch, mode, exist_ok) 

564 

565 async def unlink(self, missing_ok: bool = False) -> None: 

566 try: 

567 await to_thread.run_sync(self._path.unlink) 

568 except FileNotFoundError: 

569 if not missing_ok: 

570 raise 

571 

572 if sys.version_info >= (3, 12): 

573 

574 async def walk( 

575 self, 

576 top_down: bool = True, 

577 on_error: Callable[[OSError], object] | None = None, 

578 follow_symlinks: bool = False, 

579 ) -> AsyncIterator[tuple[Path, list[str], list[str]]]: 

580 def get_next_value() -> tuple[pathlib.Path, list[str], list[str]] | None: 

581 try: 

582 return next(gen) 

583 except StopIteration: 

584 return None 

585 

586 gen = self._path.walk(top_down, on_error, follow_symlinks) 

587 while True: 

588 value = await to_thread.run_sync(get_next_value) 

589 if value is None: 

590 return 

591 

592 root, dirs, paths = value 

593 yield Path(root), dirs, paths 

594 

595 def with_name(self, name: str) -> Path: 

596 return Path(self._path.with_name(name)) 

597 

598 def with_stem(self, stem: str) -> Path: 

599 return Path(self._path.with_name(stem + self._path.suffix)) 

600 

601 def with_suffix(self, suffix: str) -> Path: 

602 return Path(self._path.with_suffix(suffix)) 

603 

604 def with_segments(self, *pathsegments: str) -> Path: 

605 return Path(*pathsegments) 

606 

607 async def write_bytes(self, data: bytes) -> int: 

608 return await to_thread.run_sync(self._path.write_bytes, data) 

609 

610 async def write_text( 

611 self, 

612 data: str, 

613 encoding: str | None = None, 

614 errors: str | None = None, 

615 newline: str | None = None, 

616 ) -> int: 

617 # Path.write_text() does not support the "newline" parameter before Python 3.10 

618 def sync_write_text() -> int: 

619 with self._path.open( 

620 "w", encoding=encoding, errors=errors, newline=newline 

621 ) as fp: 

622 return fp.write(data) 

623 

624 return await to_thread.run_sync(sync_write_text) 

625 

626 

627PathLike.register(Path)