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

1from __future__ import annotations 

2 

3import os 

4import pathlib 

5import sys 

6from collections.abc import ( 

7 AsyncIterator, 

8 Callable, 

9 Iterable, 

10 Iterator, 

11 Sequence, 

12) 

13from dataclasses import dataclass 

14from functools import partial 

15from os import PathLike 

16from typing import ( 

17 IO, 

18 TYPE_CHECKING, 

19 Any, 

20 AnyStr, 

21 ClassVar, 

22 Final, 

23 Generic, 

24 overload, 

25) 

26 

27from .. import to_thread 

28from ..abc import AsyncResource 

29 

30if TYPE_CHECKING: 

31 from types import ModuleType 

32 

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 

43 following 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 

61 underlying file 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) -> int: 

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

106 

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

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 @overload 

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

115 

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

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

118 

119 @overload 

120 async def writelines( 

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

122 ) -> None: ... 

123 

124 @overload 

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

126 

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

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

129 

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

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

132 

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

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

135 

136 async def tell(self) -> int: 

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

138 

139 async def flush(self) -> None: 

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

141 

142 

143@overload 

144async def open_file( 

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

146 mode: OpenBinaryMode, 

147 buffering: int = ..., 

148 encoding: str | None = ..., 

149 errors: str | None = ..., 

150 newline: str | None = ..., 

151 closefd: bool = ..., 

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

153) -> AsyncFile[bytes]: ... 

154 

155 

156@overload 

157async def open_file( 

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

159 mode: OpenTextMode = ..., 

160 buffering: int = ..., 

161 encoding: str | None = ..., 

162 errors: str | None = ..., 

163 newline: str | None = ..., 

164 closefd: bool = ..., 

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

166) -> AsyncFile[str]: ... 

167 

168 

169async def open_file( 

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

171 mode: str = "r", 

172 buffering: int = -1, 

173 encoding: str | None = None, 

174 errors: str | None = None, 

175 newline: str | None = None, 

176 closefd: bool = True, 

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

178) -> AsyncFile[Any]: 

179 """ 

180 Open a file asynchronously. 

181 

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

183 

184 :return: an asynchronous file object 

185 

186 """ 

187 fp = await to_thread.run_sync( 

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

189 ) 

190 return AsyncFile(fp) 

191 

192 

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

194 """ 

195 Wrap an existing file as an asynchronous file. 

196 

197 :param file: an existing file-like object 

198 :return: an asynchronous file object 

199 

200 """ 

201 return AsyncFile(file) 

202 

203 

204@dataclass(eq=False) 

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

206 iterator: Iterator[PathLike[str]] 

207 

208 async def __anext__(self) -> Path: 

209 nextval = await to_thread.run_sync( 

210 next, self.iterator, None, abandon_on_cancel=True 

211 ) 

212 if nextval is None: 

213 raise StopAsyncIteration from None 

214 

215 return Path(nextval) 

216 

217 

218class Path: 

219 """ 

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

221 

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

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

224 interface. 

225 

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

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

228 

229 Some methods may be unavailable or have limited functionality, based on the Python 

230 version: 

231 

232 * :meth:`~pathlib.Path.copy` (available on Python 3.14 or later) 

233 * :meth:`~pathlib.Path.copy_into` (available on Python 3.14 or later) 

234 * :meth:`~pathlib.Path.from_uri` (available on Python 3.13 or later) 

235 * :meth:`~pathlib.PurePath.full_match` (available on Python 3.13 or later) 

236 * :attr:`~pathlib.Path.info` (available on Python 3.14 or later) 

237 * :meth:`~pathlib.Path.is_junction` (available on Python 3.12 or later) 

238 * :meth:`~pathlib.PurePath.match` (the ``case_sensitive`` parameter is only 

239 available on Python 3.13 or later) 

240 * :meth:`~pathlib.Path.move` (available on Python 3.14 or later) 

241 * :meth:`~pathlib.Path.move_into` (available on Python 3.14 or later) 

242 * :meth:`~pathlib.PurePath.relative_to` (the ``walk_up`` parameter is only available 

243 on Python 3.12 or later) 

244 * :meth:`~pathlib.Path.walk` (available on Python 3.12 or later) 

245 

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

247 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

275 * :meth:`~pathlib.Path.resolve` 

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

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

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

279 * :meth:`~pathlib.Path.symlink_to` 

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

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

282 * :meth:`~pathlib.Path.walk` 

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

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

285 

286 Additionally, the following methods return an async iterator yielding 

287 :class:`~.Path` objects: 

288 

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

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

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

292 """ 

293 

294 __slots__ = "_path", "__weakref__" 

295 

296 __weakref__: Any 

297 

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

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

300 

301 def __fspath__(self) -> str: 

302 return self._path.__fspath__() 

303 

304 if sys.version_info >= (3, 15): 

305 

306 def __vfspath__(self) -> str: 

307 return self._path.__vfspath__() 

308 

309 def __str__(self) -> str: 

310 return self._path.__str__() 

311 

312 def __repr__(self) -> str: 

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

314 

315 def __bytes__(self) -> bytes: 

316 return self._path.__bytes__() 

317 

318 def __hash__(self) -> int: 

319 return self._path.__hash__() 

320 

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

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

323 return self._path.__eq__(target) 

324 

325 def __lt__(self, other: pathlib.PurePath | Path) -> bool: 

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

327 return self._path.__lt__(target) 

328 

329 def __le__(self, other: pathlib.PurePath | Path) -> bool: 

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

331 return self._path.__le__(target) 

332 

333 def __gt__(self, other: pathlib.PurePath | Path) -> bool: 

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

335 return self._path.__gt__(target) 

336 

337 def __ge__(self, other: pathlib.PurePath | Path) -> bool: 

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

339 return self._path.__ge__(target) 

340 

341 def __truediv__(self, other: str | PathLike[str]) -> Path: 

342 return Path(self._path / other) 

343 

344 def __rtruediv__(self, other: str | PathLike[str]) -> Path: 

345 return Path(other) / self 

346 

347 @property 

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

349 return self._path.parts 

350 

351 @property 

352 def drive(self) -> str: 

353 return self._path.drive 

354 

355 @property 

356 def root(self) -> str: 

357 return self._path.root 

358 

359 @property 

360 def anchor(self) -> str: 

361 return self._path.anchor 

362 

363 @property 

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

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

366 

367 @property 

368 def parent(self) -> Path: 

369 return Path(self._path.parent) 

370 

371 @property 

372 def name(self) -> str: 

373 return self._path.name 

374 

375 @property 

376 def suffix(self) -> str: 

377 return self._path.suffix 

378 

379 @property 

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

381 return self._path.suffixes 

382 

383 @property 

384 def stem(self) -> str: 

385 return self._path.stem 

386 

387 async def absolute(self) -> Path: 

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

389 return Path(path) 

390 

391 def as_posix(self) -> str: 

392 return self._path.as_posix() 

393 

394 def as_uri(self) -> str: 

395 return self._path.as_uri() 

396 

397 if sys.version_info >= (3, 13): 

398 parser: ClassVar[ModuleType] = pathlib.Path.parser 

399 

400 @classmethod 

401 def from_uri(cls, uri: str) -> Path: 

402 return Path(pathlib.Path.from_uri(uri)) 

403 

404 def full_match( 

405 self, path_pattern: str, *, case_sensitive: bool | None = None 

406 ) -> bool: 

407 return self._path.full_match(path_pattern, case_sensitive=case_sensitive) 

408 

409 def match( 

410 self, path_pattern: str, *, case_sensitive: bool | None = None 

411 ) -> bool: 

412 return self._path.match(path_pattern, case_sensitive=case_sensitive) 

413 else: 

414 

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

416 return self._path.match(path_pattern) 

417 

418 if sys.version_info >= (3, 14): 

419 

420 @property 

421 def info(self) -> Any: # TODO: add return type annotation when Typeshed gets it 

422 return self._path.info 

423 

424 async def copy( 

425 self, 

426 target: str | os.PathLike[str], 

427 *, 

428 follow_symlinks: bool = True, 

429 preserve_metadata: bool = False, 

430 ) -> Path: 

431 func = partial( 

432 self._path.copy, 

433 follow_symlinks=follow_symlinks, 

434 preserve_metadata=preserve_metadata, 

435 ) 

436 return Path(await to_thread.run_sync(func, pathlib.Path(target))) 

437 

438 async def copy_into( 

439 self, 

440 target_dir: str | os.PathLike[str], 

441 *, 

442 follow_symlinks: bool = True, 

443 preserve_metadata: bool = False, 

444 ) -> Path: 

445 func = partial( 

446 self._path.copy_into, 

447 follow_symlinks=follow_symlinks, 

448 preserve_metadata=preserve_metadata, 

449 ) 

450 return Path(await to_thread.run_sync(func, pathlib.Path(target_dir))) 

451 

452 async def move(self, target: str | os.PathLike[str]) -> Path: 

453 # Upstream does not handle anyio.Path properly as a PathLike 

454 target = pathlib.Path(target) 

455 return Path(await to_thread.run_sync(self._path.move, target)) 

456 

457 async def move_into( 

458 self, 

459 target_dir: str | os.PathLike[str], 

460 ) -> Path: 

461 return Path(await to_thread.run_sync(self._path.move_into, target_dir)) 

462 

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

464 try: 

465 self.relative_to(other) 

466 return True 

467 except ValueError: 

468 return False 

469 

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

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

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

473 

474 @classmethod 

475 async def cwd(cls) -> Path: 

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

477 return cls(path) 

478 

479 async def exists(self) -> bool: 

480 return await to_thread.run_sync(self._path.exists, abandon_on_cancel=True) 

481 

482 async def expanduser(self) -> Path: 

483 return Path( 

484 await to_thread.run_sync(self._path.expanduser, abandon_on_cancel=True) 

485 ) 

486 

487 if sys.version_info < (3, 12): 

488 # Python 3.11 and earlier 

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

490 gen = self._path.glob(pattern) 

491 return _PathIterator(gen) 

492 elif (3, 12) <= sys.version_info < (3, 13): 

493 # changed in Python 3.12: 

494 # - The case_sensitive parameter was added. 

495 def glob( 

496 self, 

497 pattern: str, 

498 *, 

499 case_sensitive: bool | None = None, 

500 ) -> AsyncIterator[Path]: 

501 gen = self._path.glob(pattern, case_sensitive=case_sensitive) 

502 return _PathIterator(gen) 

503 elif sys.version_info >= (3, 13): 

504 # Changed in Python 3.13: 

505 # - The recurse_symlinks parameter was added. 

506 # - The pattern parameter accepts a path-like object. 

507 def glob( # type: ignore[misc] # mypy doesn't allow for differing signatures in a conditional block 

508 self, 

509 pattern: str | PathLike[str], 

510 *, 

511 case_sensitive: bool | None = None, 

512 recurse_symlinks: bool = False, 

513 ) -> AsyncIterator[Path]: 

514 gen = self._path.glob( 

515 pattern, # type: ignore[arg-type] 

516 case_sensitive=case_sensitive, 

517 recurse_symlinks=recurse_symlinks, 

518 ) 

519 return _PathIterator(gen) 

520 

521 async def group(self) -> str: 

522 return await to_thread.run_sync(self._path.group, abandon_on_cancel=True) 

523 

524 async def hardlink_to( 

525 self, target: str | bytes | PathLike[str] | PathLike[bytes] 

526 ) -> None: 

527 if isinstance(target, Path): 

528 target = target._path 

529 

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

531 

532 @classmethod 

533 async def home(cls) -> Path: 

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

535 return cls(home_path) 

536 

537 def is_absolute(self) -> bool: 

538 return self._path.is_absolute() 

539 

540 async def is_block_device(self) -> bool: 

541 return await to_thread.run_sync( 

542 self._path.is_block_device, abandon_on_cancel=True 

543 ) 

544 

545 async def is_char_device(self) -> bool: 

546 return await to_thread.run_sync( 

547 self._path.is_char_device, abandon_on_cancel=True 

548 ) 

549 

550 async def is_dir(self) -> bool: 

551 return await to_thread.run_sync(self._path.is_dir, abandon_on_cancel=True) 

552 

553 async def is_fifo(self) -> bool: 

554 return await to_thread.run_sync(self._path.is_fifo, abandon_on_cancel=True) 

555 

556 async def is_file(self) -> bool: 

557 return await to_thread.run_sync(self._path.is_file, abandon_on_cancel=True) 

558 

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

560 

561 async def is_junction(self) -> bool: 

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

563 

564 async def is_mount(self) -> bool: 

565 return await to_thread.run_sync( 

566 os.path.ismount, self._path, abandon_on_cancel=True 

567 ) 

568 

569 if sys.version_info < (3, 15): 

570 

571 def is_reserved(self) -> bool: 

572 return self._path.is_reserved() 

573 

574 async def is_socket(self) -> bool: 

575 return await to_thread.run_sync(self._path.is_socket, abandon_on_cancel=True) 

576 

577 async def is_symlink(self) -> bool: 

578 return await to_thread.run_sync(self._path.is_symlink, abandon_on_cancel=True) 

579 

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

581 gen = ( 

582 self._path.iterdir() 

583 if sys.version_info < (3, 13) 

584 else await to_thread.run_sync(self._path.iterdir, abandon_on_cancel=True) 

585 ) 

586 async for path in _PathIterator(gen): 

587 yield path 

588 

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

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

591 

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

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

594 

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

596 return await to_thread.run_sync(self._path.lstat, abandon_on_cancel=True) 

597 

598 async def mkdir( 

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

600 ) -> None: 

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

602 

603 @overload 

604 async def open( 

605 self, 

606 mode: OpenBinaryMode, 

607 buffering: int = ..., 

608 encoding: str | None = ..., 

609 errors: str | None = ..., 

610 newline: str | None = ..., 

611 ) -> AsyncFile[bytes]: ... 

612 

613 @overload 

614 async def open( 

615 self, 

616 mode: OpenTextMode = ..., 

617 buffering: int = ..., 

618 encoding: str | None = ..., 

619 errors: str | None = ..., 

620 newline: str | None = ..., 

621 ) -> AsyncFile[str]: ... 

622 

623 async def open( 

624 self, 

625 mode: str = "r", 

626 buffering: int = -1, 

627 encoding: str | None = None, 

628 errors: str | None = None, 

629 newline: str | None = None, 

630 ) -> AsyncFile[Any]: 

631 fp = await to_thread.run_sync( 

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

633 ) 

634 return AsyncFile(fp) 

635 

636 async def owner(self) -> str: 

637 return await to_thread.run_sync(self._path.owner, abandon_on_cancel=True) 

638 

639 async def read_bytes(self) -> bytes: 

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

641 

642 async def read_text( 

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

644 ) -> str: 

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

646 

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

648 

649 def relative_to( 

650 self, *other: str | PathLike[str], walk_up: bool = False 

651 ) -> Path: 

652 # relative_to() should work with any PathLike but it doesn't 

653 others = [pathlib.Path(other) for other in other] 

654 return Path(self._path.relative_to(*others, walk_up=walk_up)) 

655 

656 else: 

657 

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

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

660 

661 async def readlink(self) -> Path: 

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

663 return Path(target) 

664 

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

666 if isinstance(target, Path): 

667 target = target._path 

668 

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

670 return Path(target) 

671 

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

673 if isinstance(target, Path): 

674 target = target._path 

675 

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

677 return Path(target) 

678 

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

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

681 return Path(await to_thread.run_sync(func, abandon_on_cancel=True)) 

682 

683 if sys.version_info < (3, 12): 

684 # Pre Python 3.12 

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

686 gen = self._path.rglob(pattern) 

687 return _PathIterator(gen) 

688 elif (3, 12) <= sys.version_info < (3, 13): 

689 # Changed in Python 3.12: 

690 # - The case_sensitive parameter was added. 

691 def rglob( 

692 self, pattern: str, *, case_sensitive: bool | None = None 

693 ) -> AsyncIterator[Path]: 

694 gen = self._path.rglob(pattern, case_sensitive=case_sensitive) 

695 return _PathIterator(gen) 

696 elif sys.version_info >= (3, 13): 

697 # Changed in Python 3.13: 

698 # - The recurse_symlinks parameter was added. 

699 # - The pattern parameter accepts a path-like object. 

700 def rglob( # type: ignore[misc] # mypy doesn't allow for differing signatures in a conditional block 

701 self, 

702 pattern: str | PathLike[str], 

703 *, 

704 case_sensitive: bool | None = None, 

705 recurse_symlinks: bool = False, 

706 ) -> AsyncIterator[Path]: 

707 gen = self._path.rglob( 

708 pattern, # type: ignore[arg-type] 

709 case_sensitive=case_sensitive, 

710 recurse_symlinks=recurse_symlinks, 

711 ) 

712 return _PathIterator(gen) 

713 

714 async def rmdir(self) -> None: 

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

716 

717 async def samefile(self, other_path: str | PathLike[str]) -> bool: 

718 if isinstance(other_path, Path): 

719 other_path = other_path._path 

720 

721 return await to_thread.run_sync( 

722 self._path.samefile, other_path, abandon_on_cancel=True 

723 ) 

724 

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

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

727 return await to_thread.run_sync(func, self._path, abandon_on_cancel=True) 

728 

729 async def symlink_to( 

730 self, 

731 target: str | bytes | PathLike[str] | PathLike[bytes], 

732 target_is_directory: bool = False, 

733 ) -> None: 

734 if isinstance(target, Path): 

735 target = target._path 

736 

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

738 

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

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

741 

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

743 try: 

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

745 except FileNotFoundError: 

746 if not missing_ok: 

747 raise 

748 

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

750 

751 async def walk( 

752 self, 

753 top_down: bool = True, 

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

755 follow_symlinks: bool = False, 

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

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

758 try: 

759 return next(gen) 

760 except StopIteration: 

761 return None 

762 

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

764 while True: 

765 value = await to_thread.run_sync(get_next_value) 

766 if value is None: