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 def __str__(self) -> str: 

305 return self._path.__str__() 

306 

307 def __repr__(self) -> str: 

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

309 

310 def __bytes__(self) -> bytes: 

311 return self._path.__bytes__() 

312 

313 def __hash__(self) -> int: 

314 return self._path.__hash__() 

315 

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

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

318 return self._path.__eq__(target) 

319 

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

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

322 return self._path.__lt__(target) 

323 

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

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

326 return self._path.__le__(target) 

327 

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

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

330 return self._path.__gt__(target) 

331 

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

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

334 return self._path.__ge__(target) 

335 

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

337 return Path(self._path / other) 

338 

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

340 return Path(other) / self 

341 

342 @property 

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

344 return self._path.parts 

345 

346 @property 

347 def drive(self) -> str: 

348 return self._path.drive 

349 

350 @property 

351 def root(self) -> str: 

352 return self._path.root 

353 

354 @property 

355 def anchor(self) -> str: 

356 return self._path.anchor 

357 

358 @property 

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

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

361 

362 @property 

363 def parent(self) -> Path: 

364 return Path(self._path.parent) 

365 

366 @property 

367 def name(self) -> str: 

368 return self._path.name 

369 

370 @property 

371 def suffix(self) -> str: 

372 return self._path.suffix 

373 

374 @property 

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

376 return self._path.suffixes 

377 

378 @property 

379 def stem(self) -> str: 

380 return self._path.stem 

381 

382 async def absolute(self) -> Path: 

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

384 return Path(path) 

385 

386 def as_posix(self) -> str: 

387 return self._path.as_posix() 

388 

389 def as_uri(self) -> str: 

390 return self._path.as_uri() 

391 

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

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

394 

395 @classmethod 

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

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

398 

399 def full_match( 

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

401 ) -> bool: 

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

403 

404 def match( 

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

406 ) -> bool: 

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

408 else: 

409 

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

411 return self._path.match(path_pattern) 

412 

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

414 

415 @property 

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

417 return self._path.info 

418 

419 async def copy( 

420 self, 

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

422 *, 

423 follow_symlinks: bool = True, 

424 preserve_metadata: bool = False, 

425 ) -> Path: 

426 func = partial( 

427 self._path.copy, 

428 follow_symlinks=follow_symlinks, 

429 preserve_metadata=preserve_metadata, 

430 ) 

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

432 

433 async def copy_into( 

434 self, 

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

436 *, 

437 follow_symlinks: bool = True, 

438 preserve_metadata: bool = False, 

439 ) -> Path: 

440 func = partial( 

441 self._path.copy_into, 

442 follow_symlinks=follow_symlinks, 

443 preserve_metadata=preserve_metadata, 

444 ) 

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

446 

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

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

449 target = pathlib.Path(target) 

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

451 

452 async def move_into( 

453 self, 

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

455 ) -> Path: 

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

457 

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

459 try: 

460 self.relative_to(other) 

461 return True 

462 except ValueError: 

463 return False 

464 

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

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

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

468 

469 @classmethod 

470 async def cwd(cls) -> Path: 

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

472 return cls(path) 

473 

474 async def exists(self) -> bool: 

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

476 

477 async def expanduser(self) -> Path: 

478 return Path( 

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

480 ) 

481 

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

483 # Python 3.11 and earlier 

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

485 gen = self._path.glob(pattern) 

486 return _PathIterator(gen) 

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

488 # changed in Python 3.12: 

489 # - The case_sensitive parameter was added. 

490 def glob( 

491 self, 

492 pattern: str, 

493 *, 

494 case_sensitive: bool | None = None, 

495 ) -> AsyncIterator[Path]: 

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

497 return _PathIterator(gen) 

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

499 # Changed in Python 3.13: 

500 # - The recurse_symlinks parameter was added. 

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

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

503 self, 

504 pattern: str | PathLike[str], 

505 *, 

506 case_sensitive: bool | None = None, 

507 recurse_symlinks: bool = False, 

508 ) -> AsyncIterator[Path]: 

509 gen = self._path.glob( 

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

511 case_sensitive=case_sensitive, 

512 recurse_symlinks=recurse_symlinks, 

513 ) 

514 return _PathIterator(gen) 

515 

516 async def group(self) -> str: 

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

518 

519 async def hardlink_to( 

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

521 ) -> None: 

522 if isinstance(target, Path): 

523 target = target._path 

524 

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

526 

527 @classmethod 

528 async def home(cls) -> Path: 

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

530 return cls(home_path) 

531 

532 def is_absolute(self) -> bool: 

533 return self._path.is_absolute() 

534 

535 async def is_block_device(self) -> bool: 

536 return await to_thread.run_sync( 

537 self._path.is_block_device, abandon_on_cancel=True 

538 ) 

539 

540 async def is_char_device(self) -> bool: 

541 return await to_thread.run_sync( 

542 self._path.is_char_device, abandon_on_cancel=True 

543 ) 

544 

545 async def is_dir(self) -> bool: 

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

547 

548 async def is_fifo(self) -> bool: 

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

550 

551 async def is_file(self) -> bool: 

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

553 

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

555 

556 async def is_junction(self) -> bool: 

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

558 

559 async def is_mount(self) -> bool: 

560 return await to_thread.run_sync( 

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

562 ) 

563 

564 def is_reserved(self) -> bool: 

565 return self._path.is_reserved() 

566 

567 async def is_socket(self) -> bool: 

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

569 

570 async def is_symlink(self) -> bool: 

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

572 

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

574 gen = ( 

575 self._path.iterdir() 

576 if sys.version_info < (3, 13) 

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

578 ) 

579 async for path in _PathIterator(gen): 

580 yield path 

581 

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

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

584 

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

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

587 

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

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

590 

591 async def mkdir( 

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

593 ) -> None: 

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

595 

596 @overload 

597 async def open( 

598 self, 

599 mode: OpenBinaryMode, 

600 buffering: int = ..., 

601 encoding: str | None = ..., 

602 errors: str | None = ..., 

603 newline: str | None = ..., 

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

605 

606 @overload 

607 async def open( 

608 self, 

609 mode: OpenTextMode = ..., 

610 buffering: int = ..., 

611 encoding: str | None = ..., 

612 errors: str | None = ..., 

613 newline: str | None = ..., 

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

615 

616 async def open( 

617 self, 

618 mode: str = "r", 

619 buffering: int = -1, 

620 encoding: str | None = None, 

621 errors: str | None = None, 

622 newline: str | None = None, 

623 ) -> AsyncFile[Any]: 

624 fp = await to_thread.run_sync( 

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

626 ) 

627 return AsyncFile(fp) 

628 

629 async def owner(self) -> str: 

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

631 

632 async def read_bytes(self) -> bytes: 

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

634 

635 async def read_text( 

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

637 ) -> str: 

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

639 

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

641 

642 def relative_to( 

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

644 ) -> Path: 

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

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

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

648 

649 else: 

650 

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

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

653 

654 async def readlink(self) -> Path: 

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

656 return Path(target) 

657 

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

659 if isinstance(target, Path): 

660 target = target._path 

661 

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

663 return Path(target) 

664 

665 async def replace(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.replace, target) 

670 return Path(target) 

671 

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

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

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

675 

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

677 # Pre Python 3.12 

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

679 gen = self._path.rglob(pattern) 

680 return _PathIterator(gen) 

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

682 # Changed in Python 3.12: 

683 # - The case_sensitive parameter was added. 

684 def rglob( 

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

686 ) -> AsyncIterator[Path]: 

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

688 return _PathIterator(gen) 

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

690 # Changed in Python 3.13: 

691 # - The recurse_symlinks parameter was added. 

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

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

694 self, 

695 pattern: str | PathLike[str], 

696 *, 

697 case_sensitive: bool | None = None, 

698 recurse_symlinks: bool = False, 

699 ) -> AsyncIterator[Path]: 

700 gen = self._path.rglob( 

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

702 case_sensitive=case_sensitive, 

703 recurse_symlinks=recurse_symlinks, 

704 ) 

705 return _PathIterator(gen) 

706 

707 async def rmdir(self) -> None: 

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

709 

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

711 if isinstance(other_path, Path): 

712 other_path = other_path._path 

713 

714 return await to_thread.run_sync( 

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

716 ) 

717 

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

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

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

721 

722 async def symlink_to( 

723 self, 

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

725 target_is_directory: bool = False, 

726 ) -> None: 

727 if isinstance(target, Path): 

728 target = target._path 

729 

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

731 

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

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

734 

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

736 try: 

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

738 except FileNotFoundError: 

739 if not missing_ok: 

740 raise 

741 

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

743 

744 async def walk( 

745 self, 

746 top_down: bool = True, 

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

748 follow_symlinks: bool = False, 

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

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

751 try: 

752 return next(gen) 

753 except StopIteration: 

754 return None 

755 

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

757 while True: 

758 value = await to_thread.run_sync(get_next_value) 

759 if value is None: 

760 return 

761 

762 root, dirs, paths = value 

763 yield Path(root), dirs, paths 

764 

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

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