Backport types and methods to local (#421)

* upath.implementations.local: fix annotations and backport methods

* tests: test cases for .copy and .copy_into

* tests: fix test case crosspolution due to memory fs

* upath: fix py3.14+ discrepancy between __open_reader__ and __open_rb__

Author: ap--

upath/core.py

@@ -614,6 +614,11 @@ def iterdir(self) -> Iterator[Self]:
     def __open_reader__(self) -> BinaryIO:
         return self.fs.open(self.path, mode="rb")
 
+    if sys.version_info >= (3, 14):
+
+        def __open_rb__(self, buffering: int = UNSET_DEFAULT) -> BinaryIO:
+            return self.open("rb", buffering=buffering)
+
     def readlink(self) -> Self:
         _raise_unsupported(type(self).__name__, "readlink")
 

upath/implementations/local.py

@@ -8,6 +8,9 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Callable
+from typing import Literal
+from typing import overload
 from urllib.parse import SplitResult
 
 from fsspec import AbstractFileSystem
@@ -19,14 +22,27 @@
 from upath._protocol import compatible_protocol
 from upath.core import UPath
 from upath.core import _UPathMixin
+from upath.types import UNSET_DEFAULT
 from upath.types import JoinablePathLike
+from upath.types import PathInfo
+from upath.types import ReadablePath
+from upath.types import ReadablePathLike
+from upath.types import SupportsPathLike
+from upath.types import WritablePath
 
 if TYPE_CHECKING:
+    from typing import IO
+    from typing import BinaryIO
+    from typing import TextIO
+    from typing import TypeVar
+
     if sys.version_info >= (3, 11):
         from typing import Self
     else:
         from typing_extensions import Self
 
+    _WT = TypeVar("_WT", bound="WritablePath")
+
 __all__ = [
     "LocalPath",
     "PosixUPath",
@@ -66,6 +82,27 @@ def _warn_protocol_storage_options(
     )
 
 
+class _LocalPathInfo(PathInfo):
+    """Backported PathInfo implementation for LocalPath.
+    todo: currently not handling symlinks correctly.
+    """
+
+    def __init__(self, path: LocalPath) -> None:
+        self._path = path.path
+
+    def exists(self, *, follow_symlinks: bool = True) -> bool:
+        return os.path.exists(self._path)
+
+    def is_dir(self, *, follow_symlinks: bool = True) -> bool:
+        return os.path.isdir(self._path)
+
+    def is_file(self, *, follow_symlinks: bool = True) -> bool:
+        return os.path.isfile(self._path)
+
+    def is_symlink(self) -> bool:
+        return os.path.islink(self._path)
+
+
 class LocalPath(_UPathMixin, pathlib.Path):
     __slots__ = (
         "_chain",
@@ -147,6 +184,27 @@ def _init(self, **kwargs: Any) -> None:
             super()._init(**kwargs)  # type: ignore[misc]
             self._chain = Chain(ChainSegment(str(self), "", {}))
 
+    def __vfspath__(self) -> str:
+        return self.__fspath__()
+
+    def __open_reader__(self) -> BinaryIO:
+        return self.open("rb")
+
+    if sys.version_info >= (3, 14):
+
+        def __open_rb__(self, buffering: int = UNSET_DEFAULT) -> BinaryIO:
+            return self.open("rb", buffering=buffering)
+
+    def __open_writer__(self, mode: Literal["a", "w", "x"]) -> BinaryIO:
+        if mode == "w":
+            return self.open(mode="wb")
+        elif mode == "a":
+            return self.open(mode="ab")
+        elif mode == "x":
+            return self.open(mode="xb")
+        else:
+            raise ValueError(f"invalid mode: {mode}")
+
     def with_segments(self, *pathsegments: str | os.PathLike[str]) -> Self:
         return type(self)(
             *pathsegments,
@@ -190,6 +248,149 @@ def __rtruediv__(self, other) -> Self:
             else other
         )
 
+    @overload  # type: ignore[override]
+    def open(
+        self,
+        mode: Literal["r", "w", "a"] = "r",
+        buffering: int = ...,
+        encoding: str = ...,
+        errors: str = ...,
+        newline: str = ...,
+        **fsspec_kwargs: Any,
+    ) -> TextIO: ...
+
+    @overload
+    def open(
+        self,
+        mode: Literal["rb", "wb", "ab", "xb"],
+        buffering: int = ...,
+        encoding: str = ...,
+        errors: str = ...,
+        newline: str = ...,
+        **fsspec_kwargs: Any,
+    ) -> BinaryIO: ...
+
+    @overload
+    def open(
+        self,
+        mode: str,
+        buffering: int = ...,
+        encoding: str | None = ...,
+        errors: str | None = ...,
+        newline: str | None = ...,
+        **fsspec_kwargs: Any,
+    ) -> IO[Any]: ...
+
+    def open(
+        self,
+        mode: str = "r",
+        buffering: int = UNSET_DEFAULT,
+        encoding: str | None = UNSET_DEFAULT,
+        errors: str | None = UNSET_DEFAULT,
+        newline: str | None = UNSET_DEFAULT,
+        **fsspec_kwargs: Any,
+    ) -> IO[Any]:
+        if not fsspec_kwargs:
+            kwargs: dict[str, str | int | None] = {}
+            if buffering is not UNSET_DEFAULT:
+                kwargs["buffering"] = buffering
+            if encoding is not UNSET_DEFAULT:
+                kwargs["encoding"] = encoding
+            if errors is not UNSET_DEFAULT:
+                kwargs["errors"] = errors
+            if newline is not UNSET_DEFAULT:
+                kwargs["newline"] = newline
+            return super().open(mode, **kwargs)  # type: ignore  # noqa: E501
+        return UPath.open.__get__(self)(
+            mode,
+            buffering=buffering,
+            encoding=encoding,
+            errors=errors,
+            newline=newline,
+            **fsspec_kwargs,
+        )
+
+    if sys.version_info < (3, 14):
+
+        @overload
+        def copy(self, target: _WT, **kwargs: Any) -> _WT: ...
+
+        @overload
+        def copy(self, target: SupportsPathLike | str, **kwargs: Any) -> Self: ...
+
+        def copy(
+            self, target: _WT | SupportsPathLike | str, **kwargs: Any
+        ) -> _WT | Self:
+            # hacky workaround for missing pathlib.Path.copy in python < 3.14
+            # todo: revisit
+            _copy: Any = ReadablePath.copy.__get__(self)
+            if not isinstance(target, UPath):
+                return _copy(self.with_segments(str(target)), **kwargs)
+            else:
+                return _copy(target, **kwargs)
+
+        @overload
+        def copy_into(self, target_dir: _WT, **kwargs: Any) -> _WT: ...
+
+        @overload
+        def copy_into(
+            self, target_dir: SupportsPathLike | str, **kwargs: Any
+        ) -> Self: ...
+
+        def copy_into(
+            self,
+            target_dir: _WT | SupportsPathLike | str,
+            **kwargs: Any,
+        ) -> _WT | Self:
+            # hacky workaround for missing pathlib.Path.copy_into in python < 3.14
+            # todo: revisit
+            _copy_into: Any = ReadablePath.copy_into.__get__(self)
+            if not isinstance(target_dir, UPath):
+                return _copy_into(self.with_segments(str(target_dir)), **kwargs)
+            else:
+                return _copy_into(target_dir, **kwargs)
+
+        @property
+        def info(self) -> PathInfo:
+            return _LocalPathInfo(self)
+
+    if sys.version_info < (3, 13):
+
+        def full_match(self, pattern: str) -> bool:
+            # hacky workaround for missing pathlib.Path.full_match in python < 3.13
+            # todo: revisit
+            return self.match(pattern)
+
+    if sys.version_info < (3, 12):
+
+        def is_junction(self) -> bool:
+            return False
+
+        def walk(
+            self,
+            top_down: bool = True,
+            on_error: Callable[[Exception], Any] | None = None,
+            follow_symlinks: bool = False,
+        ) -> Iterator[tuple[Self, list[str], list[str]]]:
+            _walk = ReadablePath.walk.__get__(self)
+            return _walk(top_down, on_error, follow_symlinks)
+
+    if sys.version_info < (3, 10):
+
+        def hardlink_to(self, target: ReadablePathLike) -> None:
+            try:
+                os.link(target, self)
+            except AttributeError:
+                raise NotImplementedError
+
+    if not hasattr(pathlib.Path, "_copy_from"):
+
+        def _copy_from(
+            self, source: ReadablePath | LocalPath, follow_symlinks: bool = True
+        ) -> None:
+            _copy_from: Any = WritablePath._copy_from.__get__(self)
+            _copy_from(source, follow_symlinks=follow_symlinks)
+
 
 UPath.register(LocalPath)
 

upath/tests/cases.py

@@ -567,3 +567,42 @@ def test_info(self):
         assert p1.info.is_file() is False
         assert p1.info.is_dir() is True
         assert p1.info.is_symlink() is False
+
+    def test_copy_local(self, tmp_path: Path):
+        target = UPath(tmp_path) / "target-file1.txt"
+
+        source = self.path / "file1.txt"
+        content = source.read_text()
+        source.copy(target)
+        assert target.exists()
+        assert target.read_text() == content
+
+    def test_copy_into_local(self, tmp_path: Path):
+        target_dir = UPath(tmp_path) / "target-dir"
+        target_dir.mkdir()
+
+        source = self.path / "file1.txt"
+        content = source.read_text()
+        source.copy_into(target_dir)
+        target = target_dir / "file1.txt"
+        assert target.exists()
+        assert target.read_text() == content
+
+    def test_copy_memory(self, clear_fsspec_memory_cache):
+        target = UPath("memory:///target-file1.txt")
+        source = self.path / "file1.txt"
+        content = source.read_text()
+        source.copy(target)
+        assert target.exists()
+        assert target.read_text() == content
+
+    def test_copy_into_memory(self, clear_fsspec_memory_cache):
+        target_dir = UPath("memory:///target-dir")
+        target_dir.mkdir()
+
+        source = self.path / "file1.txt"
+        content = source.read_text()
+        source.copy_into(target_dir)
+        target = target_dir / "file1.txt"
+        assert target.exists()
+        assert target.read_text() == content

upath/tests/conftest.py

@@ -10,6 +10,7 @@
 
 import fsspec
 import pytest
+from fsspec import get_filesystem_class
 from fsspec.implementations.local import LocalFileSystem
 from fsspec.implementations.local import make_path_posix
 from fsspec.implementations.smb import SMBFileSystem
@@ -47,6 +48,18 @@ def clear_registry():
         _registry.clear()
 
 
+@pytest.fixture(scope="function")
+def clear_fsspec_memory_cache():
+    fs_cls = get_filesystem_class("memory")
+    pseudo_dirs = fs_cls.pseudo_dirs.copy()
+    store = fs_cls.store.copy()
+    try:
+        yield
+    finally:
+        fs_cls.pseudo_dirs = pseudo_dirs
+        fs_cls.store = store
+
+
 @pytest.fixture(scope="function")
 def local_testdir(tmp_path, clear_registry):
     folder1 = tmp_path.joinpath("folder1")
@@ -257,10 +270,10 @@ def http_server(tmp_path_factory):
     requests = pytest.importorskip("requests")
     pytest.importorskip("http.server")
     proc = subprocess.Popen(
-        shlex.split(f"python -m http.server --directory {http_tempdir} 8080")
+        shlex.split(f"python -m http.server --directory {http_tempdir} 18080")
     )
     try:
-        url = "http://127.0.0.1:8080/folder"
+        url = "http://127.0.0.1:18080/folder"
         path = Path(http_tempdir) / "folder"
         path.mkdir()
         timeout = 10

upath/tests/implementations/test_data.py

@@ -224,3 +224,43 @@ def test_info(self):
         assert p0.info.is_file() is True
         assert p0.info.is_dir() is False
         assert p0.info.is_symlink() is False
+
+    def test_copy_local(self, tmp_path):
+        target = UPath(tmp_path) / "target-file1.txt"
+
+        source = UPath("data:text/plain;base64,aGVsbG8gd29ybGQ=")
+        content = source.read_text()
+        source.copy(target)
+        assert target.exists()
+        assert target.read_text() == content
+
+    def test_copy_into_local(self, tmp_path):
+        target_dir = UPath(tmp_path) / "target-dir"
+        target_dir.mkdir()
+
+        source = UPath("data:text/plain;base64,aGVsbG8gd29ybGQ=")
+        content = source.read_text()
+        source.copy_into(target_dir)
+        target = target_dir / source.name
+        assert target.exists()
+        assert target.read_text() == content
+
+    def test_copy_memory(self, clear_fsspec_memory_cache):
+        target = UPath("memory:///target-file1.txt")
+
+        source = UPath("data:text/plain;base64,aGVsbG8gd29ybGQ=")
+        content = source.read_text()
+        source.copy(target)
+        assert target.exists()
+        assert target.read_text() == content
+
+    def test_copy_into_memory(self, clear_fsspec_memory_cache):
+        target_dir = UPath("memory:///target-dir")
+        target_dir.mkdir()
+
+        source = UPath("data:text/plain;base64,aGVsbG8gd29ybGQ=")
+        content = source.read_text()
+        source.copy_into(target_dir)
+        target = target_dir / source.name
+        assert target.exists()
+        assert target.read_text() == content