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

1from __future__ import annotations 

2 

3import os 

4import pathlib 

5import sys 

6from dataclasses import dataclass 

7from functools import partial 

8from os import PathLike 

9from typing import ( 

10 IO, 

11 TYPE_CHECKING, 

12 Any, 

13 AnyStr, 

14 AsyncIterator, 

15 Callable, 

16 Generic, 

17 Iterable, 

18 Iterator, 

19 Sequence, 

20 cast, 

21 overload, 

22) 

23 

24from .. import to_thread 

25from ..abc import AsyncResource 

26 

27if sys.version_info >= (3, 8): 

28 from typing import Final 

29else: 

30 from typing_extensions import Final 

31 

32if TYPE_CHECKING: 

33 from _typeshed import OpenBinaryMode, OpenTextMode, ReadableBuffer, WriteableBuffer 

34else: 

35 ReadableBuffer = OpenBinaryMode = OpenTextMode = WriteableBuffer = object 

36 

37 

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

39 """ 

40 An asynchronous file object. 

41 

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

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

44 

45 * read 

46 * read1 

47 * readline 

48 * readlines 

49 * readinto 

50 * readinto1 

51 * write 

52 * writelines 

53 * truncate 

54 * seek 

55 * tell 

56 * flush 

57 

58 All other methods are directly passed through. 

59 

60 This class supports the asynchronous context manager protocol which closes the underlying file 

61 at the end of the context block. 

62 

63 This class also supports asynchronous iteration:: 

64 

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

66 async for line in f: 

67 print(line) 

68 """ 

69 

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

71 self._fp: Any = fp 

72 

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

74 return getattr(self._fp, name) 

75 

76 @property 

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

78 """The wrapped file object.""" 

79 return self._fp 

80 

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

82 while True: 

83 line = await self.readline() 

84 if line: 

85 yield line 

86 else: 

87 break 

88 

89 async def aclose(self) -> None: 

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

91 

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

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

94 

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

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

97 

98 async def readline(self) -> AnyStr: 

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

100 

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

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

103 

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

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

106 

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

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

109 

110 @overload 

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

112 ... 

113 

114 @overload 

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

116 ... 

117 

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

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

120 

121 @overload 

122 async def writelines( 

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

124 ) -> None: 

125 ... 

126 

127 @overload 

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

129 ... 

130 

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

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

133 

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

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

136 

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

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

139 

140 async def tell(self) -> int: 

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

142 

143 async def flush(self) -> None: 

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

145 

146 

147@overload 

148async def open_file( 

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

150 mode: OpenBinaryMode, 

151 buffering: int = ..., 

152 encoding: str | None = ..., 

153 errors: str | None = ..., 

154 newline: str | None = ..., 

155 closefd: bool = ..., 

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

157) -> AsyncFile[bytes]: 

158 ... 

159 

160 

161@overload 

162async def open_file( 

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

164 mode: OpenTextMode = ..., 

165 buffering: int = ..., 

166 encoding: str | None = ..., 

167 errors: str | None = ..., 

168 newline: str | None = ..., 

169 closefd: bool = ..., 

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

171) -> AsyncFile[str]: 

172 ... 

173 

174 

175async def open_file( 

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

177 mode: str = "r", 

178 buffering: int = -1, 

179 encoding: str | None = None, 

180 errors: str | None = None, 

181 newline: str | None = None, 

182 closefd: bool = True, 

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

184) -> AsyncFile[Any]: 

185 """ 

186 Open a file asynchronously. 

187 

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

189 

190 :return: an asynchronous file object 

191 

192 """ 

193 fp = await to_thread.run_sync( 

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

195 ) 

196 return AsyncFile(fp) 

197 

198 

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

200 """ 

201 Wrap an existing file as an asynchronous file. 

202 

203 :param file: an existing file-like object 

204 :return: an asynchronous file object 

205 

206 """ 

207 return AsyncFile(file) 

208 

209 

210@dataclass(eq=False) 

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

212 iterator: Iterator[PathLike[str]] 

213 

214 async def __anext__(self) -> Path: 

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

216 if nextval is None: 

217 raise StopAsyncIteration from None 

218 

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

220 

221 

222class Path: 

223 """ 

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

225 

226 This class cannot be substituted for :class:`pathlib.Path` or :class:`pathlib.PurePath`, but 

227 it is compatible with the :class:`os.PathLike` interface. 

228 

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

230 deprecated :meth:`~pathlib.Path.link_to` method. 

231 

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

233 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

265 

266 Additionally, the following methods return an async iterator yielding :class:`~.Path` objects: 

267 

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

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

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

271 """ 

272 

273 __slots__ = "_path", "__weakref__" 

274 

275 __weakref__: Any 

276 

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

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

279 

280 def __fspath__(self) -> str: 

281 return self._path.__fspath__() 

282 

283 def __str__(self) -> str: 

284 return self._path.__str__() 

285 

286 def __repr__(self) -> str: 

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

288 

289 def __bytes__(self) -> bytes: 

290 return self._path.__bytes__() 

291 

292 def __hash__(self) -> int: 

293 return self._path.__hash__() 

294 

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

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

297 return self._path.__eq__(target) 

298 

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

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

301 return self._path.__lt__(target) 

302 

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

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

305 return self._path.__le__(target) 

306 

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

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

309 return self._path.__gt__(target) 

310 

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

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

313 return self._path.__ge__(target) 

314 

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

316 return Path(self._path / other) 

317 

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

319 return Path(other) / self 

320 

321 @property 

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

323 return self._path.parts 

324 

325 @property 

326 def drive(self) -> str: 

327 return self._path.drive 

328 

329 @property 

330 def root(self) -> str: 

331 return self._path.root 

332 

333 @property 

334 def anchor(self) -> str: 

335 return self._path.anchor 

336 

337 @property 

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

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

340 

341 @property 

342 def parent(self) -> Path: 

343 return Path(self._path.parent) 

344 

345 @property 

346 def name(self) -> str: 

347 return self._path.name 

348 

349 @property 

350 def suffix(self) -> str: 

351 return self._path.suffix 

352 

353 @property 

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

355 return self._path.suffixes 

356 

357 @property 

358 def stem(self) -> str: 

359 return self._path.stem 

360 

361 async def absolute(self) -> Path: 

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

363 return Path(path) 

364 

365 def as_posix(self) -> str: 

366 return self._path.as_posix() 

367 

368 def as_uri(self) -> str: 

369 return self._path.as_uri() 

370 

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

372 return self._path.match(path_pattern) 

373 

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

375 try: 

376 self.relative_to(*other) 

377 return True 

378 except ValueError: 

379 return False 

380 

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

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

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

384 

385 @classmethod 

386 async def cwd(cls) -> Path: 

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

388 return cls(path) 

389 

390 async def exists(self) -> bool: 

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

392 

393 async def expanduser(self) -> Path: 

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

395 

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

397 gen = self._path.glob(pattern) 

398 return _PathIterator(gen) 

399 

400 async def group(self) -> str: 

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

402 

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

404 if isinstance(target, Path): 

405 target = target._path 

406 

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

408 

409 @classmethod 

410 async def home(cls) -> Path: 

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

412 return cls(home_path) 

413 

414 def is_absolute(self) -> bool: 

415 return self._path.is_absolute() 

416 

417 async def is_block_device(self) -> bool: 

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

419 

420 async def is_char_device(self) -> bool: 

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

422 

423 async def is_dir(self) -> bool: 

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

425 

426 async def is_fifo(self) -> bool: 

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

428 

429 async def is_file(self) -> bool: 

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

431 

432 async def is_mount(self) -> bool: 

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

434 

435 def is_reserved(self) -> bool: 

436 return self._path.is_reserved() 

437 

438 async def is_socket(self) -> bool: 

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

440 

441 async def is_symlink(self) -> bool: 

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

443 

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

445 gen = self._path.iterdir() 

446 return _PathIterator(gen) 

447 

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

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

450 

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

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

453 

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

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

456 

457 async def mkdir( 

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

459 ) -> None: 

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

461 

462 @overload 

463 async def open( 

464 self, 

465 mode: OpenBinaryMode, 

466 buffering: int = ..., 

467 encoding: str | None = ..., 

468 errors: str | None = ..., 

469 newline: str | None = ..., 

470 ) -> AsyncFile[bytes]: 

471 ... 

472 

473 @overload 

474 async def open( 

475 self, 

476 mode: OpenTextMode = ..., 

477 buffering: int = ..., 

478 encoding: str | None = ..., 

479 errors: str | None = ..., 

480 newline: str | None = ..., 

481 ) -> AsyncFile[str]: 

482 ... 

483 

484 async def open( 

485 self, 

486 mode: str = "r", 

487 buffering: int = -1, 

488 encoding: str | None = None, 

489 errors: str | None = None, 

490 newline: str | None = None, 

491 ) -> AsyncFile[Any]: 

492 fp = await to_thread.run_sync( 

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

494 ) 

495 return AsyncFile(fp) 

496 

497 async def owner(self) -> str: 

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

499 

500 async def read_bytes(self) -> bytes: 

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

502 

503 async def read_text( 

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

505 ) -> str: 

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

507 

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

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

510 

511 async def readlink(self) -> Path: 

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

513 return Path(cast(str, target)) 

514 

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

516 if isinstance(target, Path): 

517 target = target._path 

518 

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

520 return Path(target) 

521 

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

523 if isinstance(target, Path): 

524 target = target._path 

525 

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

527 return Path(target) 

528 

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

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

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

532 

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

534 gen = self._path.rglob(pattern) 

535 return _PathIterator(gen) 

536 

537 async def rmdir(self) -> None: 

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

539 

540 async def samefile( 

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

542 ) -> bool: 

543 if isinstance(other_path, Path): 

544 other_path = other_path._path 

545 

546 return await to_thread.run_sync( 

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

548 ) 

549 

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

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

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

553 

554 async def symlink_to( 

555 self, 

556 target: str | pathlib.Path | Path, 

557 target_is_directory: bool = False, 

558 ) -> None: 

559 if isinstance(target, Path): 

560 target = target._path 

561 

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

563 

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

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

566 

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

568 try: 

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

570 except FileNotFoundError: 

571 if not missing_ok: 

572 raise 

573 

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

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

576 

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

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

579 

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

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

582 

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

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

585 

586 async def write_text( 

587 self, 

588 data: str, 

589 encoding: str | None = None, 

590 errors: str | None = None, 

591 newline: str | None = None, 

592 ) -> int: 

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

594 def sync_write_text() -> int: 

595 with self._path.open( 

596 "w", encoding=encoding, errors=errors, newline=newline 

597 ) as fp: 

598 return fp.write(data) 

599 

600 return await to_thread.run_sync(sync_write_text) 

601 

602 

603PathLike.register(Path)