diff --git a/upath/core.py b/upath/core.py index b18b65c9..a4e1127c 100644 --- a/upath/core.py +++ b/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") diff --git a/upath/implementations/local.py b/upath/implementations/local.py index d11a2656..a609202f 100644 --- a/upath/implementations/local.py +++ b/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) diff --git a/upath/tests/cases.py b/upath/tests/cases.py index 06cb35aa..fb0e90b2 100644 --- a/upath/tests/cases.py +++ b/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 diff --git a/upath/tests/conftest.py b/upath/tests/conftest.py index 67d307f3..46eb1d21 100644 --- a/upath/tests/conftest.py +++ b/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 diff --git a/upath/tests/implementations/test_data.py b/upath/tests/implementations/test_data.py index a1fd3bd1..840b93d1 100644 --- a/upath/tests/implementations/test_data.py +++ b/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