From 86182cb193a71533a7a12b8d45a6ad4103d41ccc Mon Sep 17 00:00:00 2001
From: Dave Hall <dave@etianen.com>
Date: Mon, 11 Mar 2024 21:35:12 +0000
Subject: [PATCH] `Path` refactor (#2959)
---
docs/source/reference-io.rst | 5 +
newsfragments/2959.feature.rst | 5 +
trio/__init__.py | 2 +-
trio/_path.py | 537 ++++++++++-----------------------
trio/_tests/test_exports.py | 41 +--
trio/_tests/test_path.py | 48 +--
6 files changed, 193 insertions(+), 445 deletions(-)
create mode 100644 newsfragments/2959.feature.rst
diff --git a/docs/source/reference-io.rst b/docs/source/reference-io.rst
index d0525e3..45752c1 100644
--- a/docs/source/reference-io.rst
+++ b/docs/source/reference-io.rst
@@ -631,6 +631,11 @@ Asynchronous path objects
.. autoclass:: Path
:members:
+ :inherited-members:
+
+.. autoclass:: PosixPath
+
+.. autoclass:: WindowsPath
.. _async-file-objects:
diff --git a/newsfragments/2959.feature.rst b/newsfragments/2959.feature.rst
new file mode 100644
index 0000000..ca90d56
--- /dev/null
+++ b/newsfragments/2959.feature.rst
@@ -0,0 +1,5 @@
+:class:`Path` is now a subclass of :class:`pathlib.PurePath`, allowing it to interoperate with other standard
+:mod:`pathlib` types.
+
+Instantiating :class:`Path` now returns a concrete platform-specific subclass, one of :class:`PosixPath` or
+:class:`WindowsPath`, matching the behavior of :class:`pathlib.Path`.
diff --git a/trio/__init__.py b/trio/__init__.py
index 5adf146..8fdee02 100644
--- a/trio/__init__.py
+++ b/trio/__init__.py
@@ -74,7 +74,7 @@ from ._highlevel_ssl_helpers import (
open_ssl_over_tcp_stream as open_ssl_over_tcp_stream,
serve_ssl_over_tcp as serve_ssl_over_tcp,
)
-from ._path import Path as Path
+from ._path import Path as Path, PosixPath as PosixPath, WindowsPath as WindowsPath
from ._signals import open_signal_receiver as open_signal_receiver
from ._ssl import (
NeedHandshakeError as NeedHandshakeError,
diff --git a/trio/_path.py b/trio/_path.py
index 508ad5d..eb06a6b 100644
--- a/trio/_path.py
+++ b/trio/_path.py
@@ -1,28 +1,18 @@
from __future__ import annotations
-import inspect
import os
import pathlib
import sys
-import types
-from collections.abc import Awaitable, Callable, Iterable, Sequence
-from functools import partial
+from collections.abc import Awaitable, Callable, Iterable
from io import BufferedRandom, BufferedReader, BufferedWriter, FileIO, TextIOWrapper
-from typing import (
- IO,
- TYPE_CHECKING,
- Any,
- BinaryIO,
- ClassVar,
- TypeVar,
- Union,
- cast,
- overload,
-)
-
-import trio
-from trio._file_io import AsyncIOWrapper as _AsyncIOWrapper
-from trio._util import async_wraps, final, wraps
+from functools import partial, update_wrapper
+from inspect import cleandoc
+from typing import IO, TYPE_CHECKING, Any, BinaryIO, ClassVar, TypeVar, overload
+
+from trio._file_io import AsyncIOWrapper, wrap_file
+from trio._util import final
+from trio.to_thread import run_sync
+
if TYPE_CHECKING:
from _typeshed import (
@@ -32,217 +22,103 @@ if TYPE_CHECKING:
OpenBinaryModeWriting,
OpenTextMode,
)
- from typing_extensions import Concatenate, Literal, ParamSpec, TypeAlias
+ from typing_extensions import Concatenate, Literal, ParamSpec, Self
P = ParamSpec("P")
-T = TypeVar("T")
-StrPath: TypeAlias = Union[str, "os.PathLike[str]"] # Only subscriptable in 3.9+
-
-
-# re-wrap return value from methods that return new instances of pathlib.Path
-def rewrap_path(value: T) -> T | Path:
- if isinstance(value, pathlib.Path):
- return Path(value)
- else:
- return value
+ PathT = TypeVar("PathT", bound="Path")
+ T = TypeVar("T")
-def _forward_factory(
- cls: AsyncAutoWrapperType,
- attr_name: str,
- attr: Callable[Concatenate[pathlib.Path, P], T],
-) -> Callable[Concatenate[Path, P], T | Path]:
- @wraps(attr)
- def wrapper(self: Path, *args: P.args, **kwargs: P.kwargs) -> T | Path:
- attr = getattr(self._wrapped, attr_name)
- value = attr(*args, **kwargs)
- return rewrap_path(value)
+def _wraps_async(
+ wrapped: Callable[..., Any]
+) -> Callable[[Callable[P, T]], Callable[P, Awaitable[T]]]:
+ def decorator(fn: Callable[P, T]) -> Callable[P, Awaitable[T]]:
+ async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+ return await run_sync(partial(fn, *args, **kwargs))
- # Assigning this makes inspect and therefore Sphinx show the original parameters.
- # It's not defined on functions normally though, this is a custom attribute.
- assert isinstance(wrapper, types.FunctionType)
- wrapper.__signature__ = inspect.signature(attr)
-
- return wrapper
+ update_wrapper(wrapper, wrapped)
+ assert wrapped.__doc__ is not None
+ wrapper.__doc__ = (
+ f"Like :meth:`~{wrapped.__module__}.{wrapped.__qualname__}`, but async.\n"
+ f"\n"
+ f"{cleandoc(wrapped.__doc__)}\n"
+ )
+ return wrapper
+ return decorator
-def _forward_magic(
- cls: AsyncAutoWrapperType, attr: Callable[..., T]
-) -> Callable[..., Path | T]:
- sentinel = object()
- @wraps(attr)
- def wrapper(self: Path, other: object = sentinel) -> Path | T:
- if other is sentinel:
- return attr(self._wrapped)
- if isinstance(other, cls):
- other = cast(Path, other)._wrapped
- value = attr(self._wrapped, other)
- return rewrap_path(value)
+def _wrap_method(
+ fn: Callable[Concatenate[pathlib.Path, P], T],
+) -> Callable[Concatenate[Path, P], Awaitable[T]]:
+ @_wraps_async(fn)
+ def wrapper(self: Path, /, *args: P.args, **kwargs: P.kwargs) -> T:
+ return fn(self._wrapped_cls(self), *args, **kwargs)
- assert isinstance(wrapper, types.FunctionType)
- wrapper.__signature__ = inspect.signature(attr)
return wrapper
-def iter_wrapper_factory(
- cls: AsyncAutoWrapperType, meth_name: str
-) -> Callable[Concatenate[Path, P], Awaitable[Iterable[Path]]]:
- @async_wraps(cls, cls._wraps, meth_name)
- async def wrapper(self: Path, *args: P.args, **kwargs: P.kwargs) -> Iterable[Path]:
- meth = getattr(self._wrapped, meth_name)
- func = partial(meth, *args, **kwargs)
- # Make sure that the full iteration is performed in the thread
- # by converting the generator produced by pathlib into a list
- items = await trio.to_thread.run_sync(lambda: list(func()))
- return (rewrap_path(item) for item in items)
+def _wrap_method_path(
+ fn: Callable[Concatenate[pathlib.Path, P], pathlib.Path],
+) -> Callable[Concatenate[PathT, P], Awaitable[PathT]]:
+ @_wraps_async(fn)
+ def wrapper(self: PathT, /, *args: P.args, **kwargs: P.kwargs) -> PathT:
+ return self.__class__(fn(self._wrapped_cls(self), *args, **kwargs))
return wrapper
-def thread_wrapper_factory(
- cls: AsyncAutoWrapperType, meth_name: str
-) -> Callable[Concatenate[Path, P], Awaitable[Path]]:
- @async_wraps(cls, cls._wraps, meth_name)
- async def wrapper(self: Path, *args: P.args, **kwargs: P.kwargs) -> Path:
- meth = getattr(self._wrapped, meth_name)
- func = partial(meth, *args, **kwargs)
- value = await trio.to_thread.run_sync(func)
- return rewrap_path(value)
-
+def _wrap_method_path_iterable(
+ fn: Callable[Concatenate[pathlib.Path, P], Iterable[pathlib.Path]],
+) -> Callable[Concatenate[PathT, P], Awaitable[Iterable[PathT]]]:
+ @_wraps_async(fn)
+ def wrapper(self: PathT, /, *args: P.args, **kwargs: P.kwargs) -> Iterable[PathT]:
+ return map(self.__class__, [*fn(self._wrapped_cls(self), *args, **kwargs)])
+
+ assert wrapper.__doc__ is not None
+ wrapper.__doc__ += (
+ f"\n"
+ f"This is an async method that returns a synchronous iterator, so you\n"
+ f"use it like:\n"
+ f"\n"
+ f".. code:: python\n"
+ f"\n"
+ f" for subpath in await mypath.{fn.__name__}():\n"
+ f" ...\n"
+ f"\n"
+ f".. note::\n"
+ f"\n"
+ f" The iterator is loaded into memory immediately during the initial\n"
+ f" call (see `issue #501\n"
+ f" <https://github.com/python-trio/trio/issues/501>`__ for discussion).\n"
+ )
return wrapper
+class Path(pathlib.PurePath):
+ """An async :class:`pathlib.Path` that executes blocking methods in :meth:`trio.to_thread.run_sync`.
+ Instantiating :class:`Path` returns a concrete platform-specific subclass, one of :class:`PosixPath` or
+ :class:`WindowsPath`.
+ """
-def classmethod_wrapper_factory(
- cls: AsyncAutoWrapperType, meth_name: str
-) -> classmethod: # type: ignore[type-arg]
- @async_wraps(cls, cls._wraps, meth_name)
- async def wrapper(cls: type[Path], *args: Any, **kwargs: Any) -> Path: # type: ignore[misc] # contains Any
- meth = getattr(cls._wraps, meth_name)
- func = partial(meth, *args, **kwargs)
- value = await trio.to_thread.run_sync(func)
- return rewrap_path(value)
-
- assert isinstance(wrapper, types.FunctionType)
- wrapper.__signature__ = inspect.signature(getattr(cls._wraps, meth_name))
- return classmethod(wrapper)
-
-
-class AsyncAutoWrapperType(type):
- _forwards: type
- _wraps: type
- _forward_magic: list[str]
- _wrap_iter: list[str]
- _forward: list[str]
-
- def __init__(
- cls, name: str, bases: tuple[type, ...], attrs: dict[str, object]
- ) -> None:
- super().__init__(name, bases, attrs)
-
- cls._forward = []
- type(cls).generate_forwards(cls, attrs)
- type(cls).generate_wraps(cls, attrs)
- type(cls).generate_magic(cls, attrs)
- type(cls).generate_iter(cls, attrs)
-
- def generate_forwards(cls, attrs: dict[str, object]) -> None:
- # forward functions of _forwards
- for attr_name, attr in cls._forwards.__dict__.items():
- if attr_name.startswith("_") or attr_name in attrs:
- continue
-
- if isinstance(attr, property):
- cls._forward.append(attr_name)
- elif isinstance(attr, types.FunctionType):
- wrapper = _forward_factory(cls, attr_name, attr)
- setattr(cls, attr_name, wrapper)
- else:
- raise TypeError(attr_name, type(attr))
-
- def generate_wraps(cls, attrs: dict[str, object]) -> None:
- # generate wrappers for functions of _wraps
- wrapper: classmethod | Callable[..., object] # type: ignore[type-arg]
- for attr_name, attr in cls._wraps.__dict__.items():
- # .z. exclude cls._wrap_iter
- if attr_name.startswith("_") or attr_name in attrs:
- continue
- if isinstance(attr, classmethod):
- wrapper = classmethod_wrapper_factory(cls, attr_name)
- setattr(cls, attr_name, wrapper)
- elif isinstance(attr, types.FunctionType):
- wrapper = thread_wrapper_factory(cls, attr_name)
- assert isinstance(wrapper, types.FunctionType)
- wrapper.__signature__ = inspect.signature(attr)
- setattr(cls, attr_name, wrapper)
- else:
- raise TypeError(attr_name, type(attr))
-
- def generate_magic(cls, attrs: dict[str, object]) -> None:
- # generate wrappers for magic
- for attr_name in cls._forward_magic:
- attr = getattr(cls._forwards, attr_name)
- wrapper = _forward_magic(cls, attr)
- setattr(cls, attr_name, wrapper)
-
- def generate_iter(cls, attrs: dict[str, object]) -> None:
- # generate wrappers for methods that return iterators
- wrapper: Callable[..., object]
- for attr_name, attr in cls._wraps.__dict__.items():
- if attr_name in cls._wrap_iter:
- wrapper = iter_wrapper_factory(cls, attr_name)
- assert isinstance(wrapper, types.FunctionType)
- wrapper.__signature__ = inspect.signature(attr)
- setattr(cls, attr_name, wrapper)
-
+ __slots__ = ()
-@final
-class Path(metaclass=AsyncAutoWrapperType):
- """A :class:`pathlib.Path` wrapper that executes blocking methods in
- :meth:`trio.to_thread.run_sync`.
+ _wrapped_cls: ClassVar[type[pathlib.Path]]
- """
+ def __new__(cls, *args: str | os.PathLike[str]) -> Self:
+ if cls is Path:
+ cls = WindowsPath if os.name == "nt" else PosixPath # type: ignore[assignment]
+ return super().__new__(cls, *args)
- _forward: ClassVar[list[str]]
- _wraps: ClassVar[type] = pathlib.Path
- _forwards: ClassVar[type] = pathlib.PurePath
- _forward_magic: ClassVar[list[str]] = [
- "__str__",
- "__bytes__",
- "__truediv__",
- "__rtruediv__",
- "__eq__",
- "__lt__",
- "__le__",
- "__gt__",
- "__ge__",
- "__hash__",
- ]
- _wrap_iter: ClassVar[list[str]] = ["glob", "rglob", "iterdir"]
-
- def __init__(self, *args: StrPath) -> None:
- self._wrapped = pathlib.Path(*args)
-
- # type checkers allow accessing any attributes on class instances with `__getattr__`
- # so we hide it behind a type guard forcing it to rely on the hardcoded attribute
- # list below.
- if not TYPE_CHECKING:
-
- def __getattr__(self, name):
- if name in self._forward:
- value = getattr(self._wrapped, name)
- return rewrap_path(value)
- raise AttributeError(name)
-
- def __dir__(self) -> list[str]:
- return [*super().__dir__(), *self._forward]
-
- def __repr__(self) -> str:
- return f"trio.Path({str(self)!r})"
+ @classmethod
+ @_wraps_async(pathlib.Path.cwd)
+ def cwd(cls) -> Self:
+ return cls(pathlib.Path.cwd())
- def __fspath__(self) -> str:
- return os.fspath(self._wrapped)
+ @classmethod
+ @_wraps_async(pathlib.Path.home)
+ def home(cls) -> Self:
+ return cls(pathlib.Path.home())
@overload
async def open(
@@ -252,7 +128,7 @@ class Path(metaclass=AsyncAutoWrapperType):
encoding: str | None = None,
errors: str | None = None,
newline: str | None = None,
- ) -> _AsyncIOWrapper[TextIOWrapper]:
+ ) -> AsyncIOWrapper[TextIOWrapper]:
...
@overload
@@ -263,7 +139,7 @@ class Path(metaclass=AsyncAutoWrapperType):
encoding: None = None,
errors: None = None,
newline: None = None,
- ) -> _AsyncIOWrapper[FileIO]:
+ ) -> AsyncIOWrapper[FileIO]:
...
@overload
@@ -274,7 +150,7 @@ class Path(metaclass=AsyncAutoWrapperType):
encoding: None = None,
errors: None = None,
newline: None = None,
- ) -> _AsyncIOWrapper[BufferedRandom]:
+ ) -> AsyncIOWrapper[BufferedRandom]:
...
@overload
@@ -285,7 +161,7 @@ class Path(metaclass=AsyncAutoWrapperType):
encoding: None = None,
errors: None = None,
newline: None = None,
- ) -> _AsyncIOWrapper[BufferedWriter]:
+ ) -> AsyncIOWrapper[BufferedWriter]:
...
@overload
@@ -296,7 +172,7 @@ class Path(metaclass=AsyncAutoWrapperType):
encoding: None = None,
errors: None = None,
newline: None = None,
- ) -> _AsyncIOWrapper[BufferedReader]:
+ ) -> AsyncIOWrapper[BufferedReader]:
...
@overload
@@ -307,7 +183,7 @@ class Path(metaclass=AsyncAutoWrapperType):
encoding: None = None,
errors: None = None,
newline: None = None,
- ) -> _AsyncIOWrapper[BinaryIO]:
+ ) -> AsyncIOWrapper[BinaryIO]:
...
@overload
@@ -318,173 +194,74 @@ class Path(metaclass=AsyncAutoWrapperType):
encoding: str | None = None,
errors: str | None = None,
newline: str | None = None,
- ) -> _AsyncIOWrapper[IO[Any]]:
+ ) -> AsyncIOWrapper[IO[Any]]:
...
- @wraps(pathlib.Path.open) # type: ignore[misc] # Overload return mismatch.
- async def open(self, *args: Any, **kwargs: Any) -> _AsyncIOWrapper[IO[Any]]:
- """Open the file pointed to by the path, like the :func:`trio.open_file`
- function does.
-
- """
-
- func = partial(self._wrapped.open, *args, **kwargs)
- value = await trio.to_thread.run_sync(func)
- return trio.wrap_file(value)
-
- if TYPE_CHECKING:
- # the dunders listed in _forward_magic that aren't seen otherwise
- # fmt: off
- def __bytes__(self) -> bytes: ...
- def __truediv__(self, other: StrPath) -> Path: ...
- def __rtruediv__(self, other: StrPath) -> Path: ...
- def __lt__(self, other: Path | pathlib.PurePath) -> bool: ...
- def __le__(self, other: Path | pathlib.PurePath) -> bool: ...
- def __gt__(self, other: Path | pathlib.PurePath) -> bool: ...
- def __ge__(self, other: Path | pathlib.PurePath) -> bool: ...
-
- # The following are ordered the same as in typeshed.
-
- # Properties produced by __getattr__() - all synchronous.
- @property
- def parts(self) -> tuple[str, ...]: ...
- @property
- def drive(self) -> str: ...
- @property
- def root(self) -> str: ...
- @property
- def anchor(self) -> str: ...
- @property
- def name(self) -> str: ...
- @property
- def suffix(self) -> str: ...
- @property
- def suffixes(self) -> list[str]: ...
- @property
- def stem(self) -> str: ...
- @property
- def parents(self) -> Sequence[pathlib.Path]: ... # TODO: Convert these to trio Paths?
- @property
- def parent(self) -> Path: ...
-
- # PurePath methods - synchronous.
- def as_posix(self) -> str: ...
- def as_uri(self) -> str: ...
- def is_absolute(self) -> bool: ...
- def is_reserved(self) -> bool: ...
- def match(self, path_pattern: str) -> bool: ...
- def relative_to(self, *other: StrPath) -> Path: ...
- def with_name(self, name: str) -> Path: ...
- def with_suffix(self, suffix: str) -> Path: ...
- def joinpath(self, *other: StrPath) -> Path: ...
-
- if sys.version_info >= (3, 9):
- def is_relative_to(self, *other: StrPath) -> bool: ...
- def with_stem(self, stem: str) -> Path: ...
-
- # pathlib.Path methods and properties - async.
- @classmethod
- async def cwd(self) -> Path: ...
-
- if sys.version_info >= (3, 10):
- async def stat(self, *, follow_symlinks: bool = True) -> os.stat_result: ...
- async def chmod(self, mode: int, *, follow_symlinks: bool = True) -> None: ...
- else:
- async def stat(self) -> os.stat_result: ...
- async def chmod(self, mode: int) -> None: ...
-
- async def exists(self) -> bool: ...
- async def glob(self, pattern: str) -> Iterable[Path]: ...
- async def is_dir(self) -> bool: ...
- async def is_file(self) -> bool: ...
- async def is_symlink(self) -> bool: ...
- async def is_socket(self) -> bool: ...
- async def is_fifo(self) -> bool: ...
- async def is_block_device(self) -> bool: ...
- async def is_char_device(self) -> bool: ...
- async def iterdir(self) -> Iterable[Path]: ...
- async def lchmod(self, mode: int) -> None: ...
- async def lstat(self) -> os.stat_result: ...
- async def mkdir(self, mode: int = 0o777, parents: bool = False, exist_ok: bool = False) -> None: ...
-
- if sys.platform != "win32":
- async def owner(self) -> str: ...
- async def group(self) -> str: ...
- async def is_mount(self) -> bool: ...
- if sys.version_info >= (3, 9):
- async def readlink(self) -> Path: ...
- async def rename(self, target: StrPath) -> Path: ...
- async def replace(self, target: StrPath) -> Path: ...
- async def resolve(self, strict: bool = False) -> Path: ...
- async def rglob(self, pattern: str) -> Iterable[Path]: ...
- async def rmdir(self) -> None: ...
- async def symlink_to(self, target: StrPath, target_is_directory: bool = False) -> None: ...
- if sys.version_info >= (3, 10):
- async def hardlink_to(self, target: str | pathlib.Path) -> None: ...
- async def touch(self, mode: int = 0o666, exist_ok: bool = True) -> None: ...
- async def unlink(self, missing_ok: bool = False) -> None: ...
- @classmethod
- async def home(self) -> Path: ...
- async def absolute(self) -> Path: ...
- async def expanduser(self) -> Path: ...
- async def read_bytes(self) -> bytes: ...
- async def read_text(self, encoding: str | None = None, errors: str | None = None) -> str: ...
- async def samefile(self, other_path: bytes | int | StrPath) -> bool: ...
- async def write_bytes(self, data: bytes) -> int: ...
-
- if sys.version_info >= (3, 10):
- async def write_text(
- self, data: str,
- encoding: str | None = None,
- errors: str | None = None,
- newline: str | None = None,
- ) -> int: ...
- else:
- async def write_text(
- self, data: str,
- encoding: str | None = None,
- errors: str | None = None,
- ) -> int: ...
-
- if sys.version_info < (3, 12):
- async def link_to(self, target: StrPath | bytes) -> None: ...
- if sys.version_info >= (3, 12):
- async def is_junction(self) -> bool: ...
- walk: Any # TODO
- async def with_segments(self, *pathsegments: StrPath) -> Path: ...
-
-
-Path.iterdir.__doc__ = """
- Like :meth:`~pathlib.Path.iterdir`, but async.
-
- This is an async method that returns a synchronous iterator, so you
- use it like::
-
- for subpath in await mypath.iterdir():
- ...
-
- Note that it actually loads the whole directory list into memory
- immediately, during the initial call. (See `issue #501
- <https://github.com/python-trio/trio/issues/501>`__ for discussion.)
-
-"""
-
-if sys.version_info < (3, 12):
- # Since we synthesise methods from the stdlib, this automatically will
- # have deprecation warnings, and disappear entirely in 3.12+.
- Path.link_to.__doc__ = """
- Like Python 3.8-3.11's :meth:`~pathlib.Path.link_to`, but async.
-
- :deprecated: This method was deprecated in Python 3.10 and entirely \
- removed in 3.12. Use :meth:`hardlink_to` instead which has \
- a more meaningful parameter order.
-"""
-
-# The value of Path.absolute.__doc__ makes a reference to
-# :meth:~pathlib.Path.absolute, which does not exist. Removing this makes more
-# sense than inventing our own special docstring for this.
-del Path.absolute.__doc__
-
-# TODO: This is likely not supported by all the static tools out there, see discussion in
-# https://github.com/python-trio/trio/pull/2631#discussion_r1185612528
-os.PathLike.register(Path)
+ @_wraps_async(pathlib.Path.open) # type: ignore[misc] # Overload return mismatch.
+ def open(self, *args: Any, **kwargs: Any) -> AsyncIOWrapper[IO[Any]]:
+ return wrap_file(self._wrapped_cls(self).open(*args, **kwargs))
+
+ def __repr__(self) -> str:
+ return f"trio.Path({str(self)!r})"
+
+ stat = _wrap_method(pathlib.Path.stat)
+ chmod = _wrap_method(pathlib.Path.chmod)
+ exists = _wrap_method(pathlib.Path.exists)
+ glob = _wrap_method_path_iterable(pathlib.Path.glob)
+ rglob = _wrap_method_path_iterable(pathlib.Path.rglob)
+ is_dir = _wrap_method(pathlib.Path.is_dir)
+ is_file = _wrap_method(pathlib.Path.is_file)
+ is_symlink = _wrap_method(pathlib.Path.is_symlink)
+ is_socket = _wrap_method(pathlib.Path.is_socket)
+ is_fifo = _wrap_method(pathlib.Path.is_fifo)
+ is_block_device = _wrap_method(pathlib.Path.is_block_device)
+ is_char_device = _wrap_method(pathlib.Path.is_char_device)
+ if sys.version_info >= (3, 12):
+ is_junction = _wrap_method(pathlib.Path.is_junction)
+ iterdir = _wrap_method_path_iterable(pathlib.Path.iterdir)
+ lchmod = _wrap_method(pathlib.Path.lchmod)
+ lstat = _wrap_method(pathlib.Path.lstat)
+ mkdir = _wrap_method(pathlib.Path.mkdir)
+ if sys.platform != "win32":
+ owner = _wrap_method(pathlib.Path.owner)
+ group = _wrap_method(pathlib.Path.group)
+ if sys.platform != "win32" or sys.version_info >= (3, 12):
+ is_mount = _wrap_method(pathlib.Path.is_mount)
+ if sys.version_info >= (3, 9):
+ readlink = _wrap_method_path(pathlib.Path.readlink)
+ rename = _wrap_method_path(pathlib.Path.rename)
+ replace = _wrap_method_path(pathlib.Path.replace)
+ resolve = _wrap_method_path(pathlib.Path.resolve)
+ rmdir = _wrap_method(pathlib.Path.rmdir)
+ symlink_to = _wrap_method(pathlib.Path.symlink_to)
+ if sys.version_info >= (3, 10):
+ hardlink_to = _wrap_method(pathlib.Path.hardlink_to)
+ touch = _wrap_method(pathlib.Path.touch)
+ unlink = _wrap_method(pathlib.Path.unlink)
+ absolute = _wrap_method_path(pathlib.Path.absolute)
+ expanduser = _wrap_method_path(pathlib.Path.expanduser)
+ read_bytes = _wrap_method(pathlib.Path.read_bytes)
+ read_text = _wrap_method(pathlib.Path.read_text)
+ samefile = _wrap_method(pathlib.Path.samefile)
+ write_bytes = _wrap_method(pathlib.Path.write_bytes)
+ write_text = _wrap_method(pathlib.Path.write_text)
+ if sys.version_info < (3, 12):
+ link_to = _wrap_method(pathlib.Path.link_to)
+
+
+@final
+class PosixPath(Path, pathlib.PurePosixPath):
+ """An async :class:`pathlib.PosixPath` that executes blocking methods in :meth:`trio.to_thread.run_sync`."""
+
+ __slots__ = ()
+
+ _wrapped_cls: ClassVar[type[pathlib.Path]] = pathlib.PosixPath
+
+
+@final
+class WindowsPath(Path, pathlib.PureWindowsPath):
+ """An async :class:`pathlib.WindowsPath` that executes blocking methods in :meth:`trio.to_thread.run_sync`."""
+
+ __slots__ = ()
+
+ _wrapped_cls: ClassVar[type[pathlib.Path]] = pathlib.WindowsPath
diff --git a/trio/_tests/test_exports.py b/trio/_tests/test_exports.py
index 7b38137..e1ced60 100644
--- a/trio/_tests/test_exports.py
+++ b/trio/_tests/test_exports.py
@@ -11,7 +11,7 @@ import socket as stdlib_socket
import sys
import types
from collections.abc import Iterator
-from pathlib import Path
+from pathlib import Path, PurePath
from types import ModuleType
from typing import Iterable, Protocol
@@ -333,7 +333,8 @@ def test_static_tool_sees_class_members(
mod_cache = next_cache / "__init__.data.json"
else:
mod_cache = mod_cache / (modname[-1] + ".data.json")
-
+ elif mod_cache.is_dir():
+ mod_cache /= "__init__.data.json"
with mod_cache.open() as f:
return json.loads(f.read())["names"][name] # type: ignore[no-any-return]
@@ -480,12 +481,6 @@ def test_static_tool_sees_class_members(
extra -= EXTRAS[class_]
assert len(extra) == before - len(EXTRAS[class_])
- # probably an issue with mypy....
- if tool == "mypy" and class_ == trio.Path and sys.platform == "win32":
- before = len(missing)
- missing -= {"owner", "group", "is_mount"}
- assert len(missing) == before - 3
-
# TODO: why is this? Is it a problem?
# see https://github.com/python-trio/trio/pull/2631#discussion_r1185615916
if class_ == trio.StapledStream:
@@ -508,25 +503,14 @@ def test_static_tool_sees_class_members(
missing.remove("__aiter__")
missing.remove("__anext__")
- # __getattr__ is intentionally hidden behind type guard. That hook then
- # forwards property accesses to PurePath, meaning these names aren't directly on
- # the class.
- if class_ == trio.Path:
- missing.remove("__getattr__")
- before = len(extra)
- extra -= {
- "anchor",
- "drive",
- "name",
- "parent",
- "parents",
- "parts",
- "root",
- "stem",
- "suffix",
- "suffixes",
- }
- assert len(extra) == before - 10
+ if class_ in (trio.Path, trio.WindowsPath, trio.PosixPath):
+ # These are from inherited subclasses.
+ missing -= PurePath.__dict__.keys()
+ # These are unix-only.
+ if tool == "mypy" and sys.platform == "win32":
+ missing -= {"owner", "is_mount", "group"}
+ if tool == "jedi" and sys.platform == "win32":
+ extra -= {"owner", "is_mount", "group"}
if missing or extra: # pragma: no cover
errors[f"{module_name}.{class_name}"] = {
@@ -588,6 +572,9 @@ def test_classes_are_final() -> None:
continue
# ... insert other special cases here ...
+ # The `Path` class needs to support inheritance to allow `WindowsPath` and `PosixPath`.
+ if class_ is trio.Path:
+ continue
# don't care about the *Statistics classes
if name.endswith("Statistics"):
continue
diff --git a/trio/_tests/test_path.py b/trio/_tests/test_path.py
index 30158f8..3528053 100644
--- a/trio/_tests/test_path.py
+++ b/trio/_tests/test_path.py
@@ -3,13 +3,12 @@ from __future__ import annotations
import os
import pathlib
from collections.abc import Awaitable, Callable
-from typing import Any, Type, Union
+from typing import Type, Union
import pytest
import trio
from trio._file_io import AsyncIOWrapper
-from trio._path import AsyncAutoWrapperType as WrapperType
@pytest.fixture
@@ -26,6 +25,16 @@ def method_pair(
return getattr(sync_path, method_name), getattr(async_path, method_name)
+@pytest.mark.skipif(os.name == "nt", reason="OS is not posix")
+async def test_instantiate_posix() -> None:
+ assert isinstance(trio.Path(), trio.PosixPath)
+
+
+@pytest.mark.skipif(os.name != "nt", reason="OS is not Windows")
+async def test_instantiate_windows() -> None:
+ assert isinstance(trio.Path(), trio.WindowsPath)
+
+
async def test_open_is_async_context_manager(path: trio.Path) -> None:
async with await path.open("w") as f:
assert isinstance(f, AsyncIOWrapper)
@@ -166,41 +175,6 @@ async def test_repr() -> None:
assert repr(path) == "trio.Path('.')"
-class MockWrapped:
- unsupported = "unsupported"
- _private = "private"
-
-
-class _MockWrapper:
- _forwards = MockWrapped
- _wraps = MockWrapped
-
-
-MockWrapper: Any = _MockWrapper # Disable type checking, it's a mock.
-
-
-async def test_type_forwards_unsupported() -> None:
- with pytest.raises(TypeError):
- WrapperType.generate_forwards(MockWrapper, {})
-
-
-async def test_type_wraps_unsupported() -> None:
- with pytest.raises(TypeError):
- WrapperType.generate_wraps(MockWrapper, {})
-
-
-async def test_type_forwards_private() -> None:
- WrapperType.generate_forwards(MockWrapper, {"unsupported": None})
-
- assert not hasattr(MockWrapper, "_private")
-
-
-async def test_type_wraps_private() -> None:
- WrapperType.generate_wraps(MockWrapper, {"unsupported": None})
-
- assert not hasattr(MockWrapper, "_private")
-
-
@pytest.mark.parametrize("meth", [trio.Path.__init__, trio.Path.joinpath])
async def test_path_wraps_path(
path: trio.Path,
--
2.45.0