codekingpro/portable-devtools

1from __future__ import annotations2 3import os4import pathlib5import sys6from collections.abc import (7    AsyncIterator,8    Callable,9    Iterable,10    Iterator,11    Sequence,12)13from dataclasses import dataclass14from functools import partial15from os import PathLike16from typing import (17    IO,18    TYPE_CHECKING,19    Any,20    AnyStr,21    ClassVar,22    Final,23    Generic,24    overload,25)26 27from .. import to_thread28from ..abc import AsyncResource29 30if TYPE_CHECKING:31    from types import ModuleType32 33    from _typeshed import OpenBinaryMode, OpenTextMode, ReadableBuffer, WriteableBuffer34else:35    ReadableBuffer = OpenBinaryMode = OpenTextMode = WriteableBuffer = object36 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 the43    following blocking methods (where available on the original file object):44 45    * read46    * read147    * readline48    * readlines49    * readinto50    * readinto151    * write52    * writelines53    * truncate54    * seek55    * tell56    * flush57 58    All other methods are directly passed through.59 60    This class supports the asynchronous context manager protocol which closes the61    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 = fp72 73    def __getattr__(self, name: str) -> object:74        return getattr(self._fp, name)75 76    @property77    def wrapped(self) -> IO[AnyStr]:78        """The wrapped file object."""79        return self._fp80 81    async def __aiter__(self) -> AsyncIterator[AnyStr]:82        while True:83            line = await self.readline()84            if line:85                yield line86            else:87                break88 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    @overload111    async def write(self: AsyncFile[bytes], b: ReadableBuffer) -> int: ...112 113    @overload114    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    @overload120    async def writelines(121        self: AsyncFile[bytes], lines: Iterable[ReadableBuffer]122    ) -> None: ...123 124    @overload125    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@overload144async 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@overload157async 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 object185 186    """187    fp = await to_thread.run_sync(188        open, file, mode, buffering, encoding, errors, newline, closefd, opener189    )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 object198    :return: an asynchronous file object199 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=True211        )212        if nextval is None:213            raise StopAsyncIteration from None214 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` or223    :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 for227    the deprecated :meth:`~pathlib.Path.link_to` method.228 229    Some methods may be unavailable or have limited functionality, based on the Python230    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 only239      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 available243      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 yielding287    :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__: Any297 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 other323        return self._path.__eq__(target)324 325    def __lt__(self, other: pathlib.PurePath | Path) -> bool:326        target = other._path if isinstance(other, Path) else other327        return self._path.__lt__(target)328 329    def __le__(self, other: pathlib.PurePath | Path) -> bool:330        target = other._path if isinstance(other, Path) else other331        return self._path.__le__(target)332 333    def __gt__(self, other: pathlib.PurePath | Path) -> bool:334        target = other._path if isinstance(other, Path) else other335        return self._path.__gt__(target)336 337    def __ge__(self, other: pathlib.PurePath | Path) -> bool:338        target = other._path if isinstance(other, Path) else other339        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) / self346 347    @property348    def parts(self) -> tuple[str, ...]:349        return self._path.parts350 351    @property352    def drive(self) -> str:353        return self._path.drive354 355    @property356    def root(self) -> str:357        return self._path.root358 359    @property360    def anchor(self) -> str:361        return self._path.anchor362 363    @property364    def parents(self) -> Sequence[Path]:365        return tuple(Path(p) for p in self._path.parents)366 367    @property368    def parent(self) -> Path:369        return Path(self._path.parent)370 371    @property372    def name(self) -> str:373        return self._path.name374 375    @property376    def suffix(self) -> str:377        return self._path.suffix378 379    @property380    def suffixes(self) -> list[str]:381        return self._path.suffixes382 383    @property384    def stem(self) -> str:385        return self._path.stem386 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.parser399 400        @classmethod401        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 = None406        ) -> 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 = None411        ) -> 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        @property421        def info(self) -> Any:  # TODO: add return type annotation when Typeshed gets it422            return self._path.info423 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 PathLike454            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 True467        except ValueError:468            return False469 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    @classmethod475    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 earlier489        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 block508            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._path529 530        await to_thread.run_sync(os.link, target, self)531 532    @classmethod533    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=True543        )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=True548        )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=True567        )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 path588 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 = False600    ) -> None:601        await to_thread.run_sync(self._path.mkdir, mode, parents, exist_ok)602 603    @overload604    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    @overload614    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, newline633        )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 = None644    ) -> 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 = False651        ) -> Path:652            # relative_to() should work with any PathLike but it doesn't653            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._path668 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._path675 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.12685        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 = None693        ) -> 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 block701            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._path720 721        return await to_thread.run_sync(722            self._path.samefile, other_path, abandon_on_cancel=True723        )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._path736 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                raise748 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 None762 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:767                    return768 769                root, dirs, paths = value770                yield Path(root), dirs, paths771 772    def with_name(self, name: str) -> Path:773        return Path(self._path.with_name(name))774 775    def with_stem(self, stem: str) -> Path:776        return Path(self._path.with_name(stem + self._path.suffix))777 778    def with_suffix(self, suffix: str) -> Path:779        return Path(self._path.with_suffix(suffix))780 781    def with_segments(self, *pathsegments: str | PathLike[str]) -> Path:782        return Path(*pathsegments)783 784    async def write_bytes(self, data: bytes) -> int:785        return await to_thread.run_sync(self._path.write_bytes, data)786 787    async def write_text(788        self,789        data: str,790        encoding: str | None = None,791        errors: str | None = None,792        newline: str | None = None,793    ) -> int:794        return await to_thread.run_sync(795            self._path.write_text, data, encoding, errors, newline796        )797 798 799PathLike.register(Path)800