diff --git a/dev/fsspec_inspector/generate_flavours.py b/dev/fsspec_inspector/generate_flavours.py index 8a6b0a56..324f007a 100644 --- a/dev/fsspec_inspector/generate_flavours.py +++ b/dev/fsspec_inspector/generate_flavours.py @@ -105,7 +105,6 @@ def __init_subclass__(cls: Any, **kwargs): "dir", "blockcache", "cached", - "simplecache", "filecache", ] @@ -116,6 +115,7 @@ def __init_subclass__(cls: Any, **kwargs): FIX_METHODS = { "GCSFileSystem": ["_strip_protocol", "_get_kwargs_from_urls", "_split_path"], + "SimpleCacheFileSystem": [], } diff --git a/upath/_chain.py b/upath/_chain.py index 63c7ffc3..83d3db5a 100644 --- a/upath/_chain.py +++ b/upath/_chain.py @@ -4,23 +4,28 @@ import warnings from collections import defaultdict from collections import deque +from collections.abc import Iterator from collections.abc import MutableMapping from collections.abc import Sequence from collections.abc import Set +from itertools import zip_longest from typing import TYPE_CHECKING from typing import Any from typing import NamedTuple +from upath._flavour import WrappedFileSystemFlavour +from upath._protocol import get_upath_protocol +from upath.registry import available_implementations +from upath.types import UNSET_DEFAULT + if TYPE_CHECKING: if sys.version_info >= (3, 11): + from typing import Never from typing import Self else: + from typing_extensions import Never from typing_extensions import Self -from upath._flavour import WrappedFileSystemFlavour -from upath._protocol import get_upath_protocol -from upath.registry import available_implementations - __all__ = [ "ChainSegment", "Chain", @@ -153,6 +158,23 @@ def nest(self) -> ChainSegment: return ChainSegment(urlpath, protocol, inkwargs) +def _iter_fileobject_protocol_options( + fileobject: str | None, + protocol: str, + storage_options: dict[str, Any], + /, +) -> Iterator[tuple[str | None, str, dict[str, Any]]]: + """yields fileobject, protocol and remaining storage options""" + so = storage_options.copy() + while "target_protocol" in so: + t_protocol = so.pop("target_protocol", "") + t_fileobject = so.pop("fo", None) # codespell:ignore fo + t_so = so.pop("target_options", {}) + yield fileobject, protocol, so + fileobject, protocol, so = t_fileobject, t_protocol, t_so + yield fileobject, protocol, so + + class FSSpecChainParser: """parse an fsspec chained urlpath""" @@ -160,7 +182,15 @@ def __init__(self) -> None: self.link: str = "::" self.known_protocols: Set[str] = set() - def unchain(self, path: str, kwargs: dict[str, Any]) -> list[ChainSegment]: + def unchain( + self, + path: str, + _deprecated_storage_options: Never = UNSET_DEFAULT, + /, + *, + protocol: str | None = None, + storage_options: dict[str, Any] | None = None, + ) -> list[ChainSegment]: """implements same behavior as fsspec.core._un_chain two differences: @@ -168,59 +198,89 @@ def unchain(self, path: str, kwargs: dict[str, Any]) -> list[ChainSegment]: 2. it checks against the known protocols for exact matches """ - # TODO: upstream to fsspec - first_bit_protocol: str | None = kwargs.pop("protocol", None) - it_bits = iter(path.split(self.link)) - bits: list[str] - if first_bit_protocol is not None: - bits = [next(it_bits)] - else: - bits = [] - for p in it_bits: - if "://" in p: # uri-like, fast-path - bits.append(p) - elif "/" in p: # path-like, fast-path - bits.append(p) - elif p in self.known_protocols: # exact match a fsspec protocol - bits.append(f"{p}://") - elif p in (m := set(available_implementations(fallback=True))): - self.known_protocols = m - bits.append(f"{p}://") - else: - bits.append(p) - - # [[url, protocol, kwargs], ...] - out: list[ChainSegment] = [] - previous_bit: str | None = None - kwargs = kwargs.copy() - first_bit_idx = len(bits) - 1 - for idx, bit in enumerate(reversed(bits)): - if idx == first_bit_idx: - protocol = first_bit_protocol or get_upath_protocol(bit) or "" - else: - protocol = get_upath_protocol(bit) or "" - flavour = WrappedFileSystemFlavour.from_protocol(protocol) - extra_kwargs = flavour.get_kwargs_from_url(bit) - kws = kwargs.pop(protocol, {}) - if bit is bits[0]: - kws.update(kwargs) - kw = dict(**extra_kwargs) - kw.update(kws) - if "target_protocol" in kw: - kw.setdefault("target_options", {}) - bit = flavour.strip_protocol(bit) or flavour.root_marker + if _deprecated_storage_options is not UNSET_DEFAULT: + warnings.warn( + "passing storage_options as positional argument is deprecated, " + "pass as keyword argument instead", + DeprecationWarning, + stacklevel=2, + ) + if storage_options is not None: + raise ValueError( + "cannot pass storage_options both positionally and as keyword" + ) + storage_options = _deprecated_storage_options + protocol = protocol or storage_options.get("protocol") + if storage_options is None: + storage_options = {} + + segments: list[ChainSegment] = [] + path_bit: str | None + next_path_overwrite: str | None = None + for proto0, bit in zip_longest([protocol], path.split(self.link)): + # get protocol and path_bit if ( - protocol in {"blockcache", "filecache", "simplecache"} - and "target_protocol" not in kw + "://" in bit # uri-like, fast-path (redundant) + or "/" in bit # path-like, fast-path ): - out.append(ChainSegment(None, protocol, kw)) - if previous_bit is not None: - bit = previous_bit + proto = get_upath_protocol(bit, protocol=proto0) + flavour = WrappedFileSystemFlavour.from_protocol(proto) + path_bit = flavour.strip_protocol(bit) + extra_so = flavour.get_kwargs_from_url(bit) + elif bit in self.known_protocols and ( + proto0 is None or bit == proto0 + ): # exact match a fsspec protocol + proto = bit + path_bit = "" + extra_so = {} + elif bit in (m := set(available_implementations(fallback=True))) and ( + proto0 is None or bit == proto0 + ): + self.known_protocols = m + proto = bit + path_bit = "" + extra_so = {} + else: + proto = get_upath_protocol(bit, protocol=proto0) + flavour = WrappedFileSystemFlavour.from_protocol(proto) + path_bit = flavour.strip_protocol(bit) + extra_so = flavour.get_kwargs_from_url(bit) + if proto in {"blockcache", "filecache", "simplecache"}: + if path_bit: + next_path_overwrite = path_bit + path_bit = None + elif next_path_overwrite is not None: + path_bit = next_path_overwrite + next_path_overwrite = None + segments.append(ChainSegment(path_bit, proto, extra_so)) + + root_so = segments[0].storage_options + for segment, proto_fo_so in zip_longest( + segments, + _iter_fileobject_protocol_options( + path_bit if segments else None, + protocol or "", + storage_options, + ), + ): + t_fo, t_proto, t_so = proto_fo_so or (None, "", {}) + if segment is None: + if next_path_overwrite is not None: + t_fo = next_path_overwrite + next_path_overwrite = None + segments.append(ChainSegment(t_fo, t_proto, t_so)) else: - out.append(ChainSegment(bit, protocol, kw)) - previous_bit = bit - out.reverse() - return out + proto = segment.protocol + # check if protocol is consistent with storage options + if t_proto and t_proto != proto: + raise ValueError( + f"protocol {proto!r} collides with target_protocol {t_proto!r}" + ) + # update the storage_options + segment.storage_options.update(root_so.pop(proto, {})) + segment.storage_options.update(t_so) + + return segments def chain(self, segments: Sequence[ChainSegment]) -> tuple[str, dict[str, Any]]: """returns a chained urlpath from the segments""" @@ -268,7 +328,7 @@ def chain(self, segments: Sequence[ChainSegment]) -> tuple[str, dict[str, Any]]: chained_kw = {"zip": {"allowZip64": False}} print(chained_path, chained_kw) out0 = _un_chain(chained_path, chained_kw) - out1 = FSSpecChainParser().unchain(chained_path, chained_kw) + out1 = FSSpecChainParser().unchain(chained_path, storage_options=chained_kw) pp(out0) pp(out1) diff --git a/upath/_flavour_sources.py b/upath/_flavour_sources.py index e5ba196c..396d6017 100644 --- a/upath/_flavour_sources.py +++ b/upath/_flavour_sources.py @@ -23,7 +23,6 @@ # - cached # - dir # - filecache -# - simplecache # protocol import errors: # - gdrive (Please install gdrive_fs for access to Google Drive) # - generic (GenericFileSystem: '_strip_protocol' not a classmethod) @@ -926,6 +925,15 @@ def _get_kwargs_from_urls(path): return out +class SimpleCacheFileSystemFlavour(AbstractFileSystemFlavour): + __orig_class__ = 'fsspec.implementations.cached.SimpleCacheFileSystem' + __orig_version__ = '2025.9.0' + protocol = ('simplecache',) + root_marker = '' + sep = '/' + local_file = True + + class TarFileSystemFlavour(AbstractFileSystemFlavour): __orig_class__ = 'fsspec.implementations.tar.TarFileSystem' __orig_version__ = '2025.9.0' diff --git a/upath/core.py b/upath/core.py index ce309265..1a1ae198 100644 --- a/upath/core.py +++ b/upath/core.py @@ -1,6 +1,6 @@ from __future__ import annotations -import os +import posixpath import sys import warnings from abc import ABCMeta @@ -423,9 +423,12 @@ def __init__( str_args0 = "." segments = chain_parser.unchain( - str_args0, {"protocol": protocol, **storage_options} + str_args0, + protocol=protocol, + storage_options=storage_options, ) - chain = Chain.from_list(segments) + # FIXME: normalization needs to happen in unchain already... + chain = Chain.from_list(Chain.from_list(segments).to_list()) if len(args) > 1: chain = chain.replace( path=WrappedFileSystemFlavour.from_protocol(protocol).join( @@ -1102,14 +1105,14 @@ def group(self) -> str: def absolute(self) -> Self: if self._relative_base is not None: - return self.cwd().joinpath(str(self)) + return self.cwd().joinpath(self.__vfspath__()) return self def is_absolute(self) -> bool: if self._relative_base is not None: return False else: - return self.parser.isabs(str(self)) + return self.parser.isabs(self.__vfspath__()) def __eq__(self, other: object) -> bool: """UPaths are considered equal if their protocol, path and @@ -1223,22 +1226,24 @@ def rename( maxdepth: int | None = UNSET_DEFAULT, **kwargs: Any, ) -> Self: - if isinstance(target, str) and self.storage_options: - target = UPath(target, **self.storage_options) + target_protocol = get_upath_protocol(target) + if target_protocol and target_protocol != self.protocol: + raise ValueError( + f"expected protocol {self.protocol!r}, got: {target_protocol!r}" + ) + if not isinstance(target, UPath): + target = str(target) + if target_protocol or (self.anchor and target.startswith(self.anchor)): + target = self.with_segments(target) + else: + target = UPath(target) if target == self: return self if self._relative_base is not None: self = self.absolute() target_protocol = get_upath_protocol(target) if target_protocol: - if target_protocol != self.protocol: - raise ValueError( - f"expected protocol {self.protocol!r}, got: {target_protocol!r}" - ) - if not isinstance(target, UPath): - target_ = UPath(target, **self.storage_options) - else: - target_ = target + target_ = target # avoid calling .resolve for subclasses of UPath if ".." in target_.parts or "." in target_.parts: target_ = target_.resolve() @@ -1247,7 +1252,7 @@ def rename( # avoid calling .resolve for subclasses of UPath if ".." in parent.parts or "." in parent.parts: parent = parent.resolve() - target_ = parent.joinpath(os.path.normpath(str(target))) + target_ = parent.joinpath(posixpath.normpath(target.path)) if recursive is not UNSET_DEFAULT: kwargs["recursive"] = recursive if maxdepth is not UNSET_DEFAULT: @@ -1275,14 +1280,20 @@ def root(self) -> str: return self.parser.splitroot(str(self))[1] def __reduce__(self): - args = tuple(self._raw_urlpaths) - kwargs = { - "protocol": self._protocol, - **self._storage_options, - } - # Include _relative_base in the state if it's set - if self._relative_base is not None: - kwargs["_relative_base"] = self._relative_base + if self._relative_base is None: + args = (self.__vfspath__(),) + kwargs = { + "protocol": self._protocol, + **self._storage_options, + } + else: + args = (self._relative_base, self.__vfspath__()) + # Include _relative_base in the state if it's set + kwargs = { + "protocol": self._protocol, + **self._storage_options, + "_relative_base": self._relative_base, + } return _make_instance, (type(self), args, kwargs) @classmethod diff --git a/upath/implementations/cached.py b/upath/implementations/cached.py index de01fb3e..0da528a6 100644 --- a/upath/implementations/cached.py +++ b/upath/implementations/cached.py @@ -1,19 +1,27 @@ from __future__ import annotations import sys +from types import MappingProxyType from typing import TYPE_CHECKING from upath.core import UPath from upath.types import JoinablePathLike if TYPE_CHECKING: + from collections.abc import Iterator + from collections.abc import Mapping + from typing import Any from typing import Literal if sys.version_info >= (3, 11): + from typing import Self from typing import Unpack else: + from typing_extensions import Self from typing_extensions import Unpack + from fsspec import AbstractFileSystem + from upath._chain import FSSpecChainParser from upath.types.storage_options import SimpleCacheStorageOptions @@ -33,3 +41,29 @@ def __init__( chain_parser: FSSpecChainParser = ..., **storage_options: Unpack[SimpleCacheStorageOptions], ) -> None: ... + + @classmethod + def _fs_factory( + cls, + urlpath: str, + protocol: str, + storage_options: Mapping[str, Any], + ) -> AbstractFileSystem: + so = dict(storage_options) + so.pop("fo", None) + return super()._fs_factory( + urlpath, + protocol, + so, + ) + + @property + def storage_options(self) -> Mapping[str, Any]: + so = self._storage_options.copy() + so.pop("fo", None) + return MappingProxyType(so) + + def iterdir(self) -> Iterator[Self]: + if self.is_file(): + raise NotADirectoryError(str(self)) + yield from super().iterdir() diff --git a/upath/tests/cases.py b/upath/tests/cases.py index 8b6e4fc0..56bb0141 100644 --- a/upath/tests/cases.py +++ b/upath/tests/cases.py @@ -1,6 +1,5 @@ import os import pickle -import re import stat import sys import warnings @@ -269,7 +268,7 @@ def test_rename(self): assert not upath.exists() assert moved.exists() # reverse with an absolute path as str - back = moved.rename(str(upath)) + back = moved.rename(upath.path) assert back == upath assert not moved.exists() assert back.exists() @@ -380,22 +379,21 @@ def test_fsspec_compat(self): fs = self.path.fs content = b"a,b,c\n1,2,3\n4,5,6" - def strip_scheme(path): - root = "" if sys.platform.startswith("win") else "/" - return root + re.sub("^[a-z0-9]+:/*", "", str(path)) - upath1 = self.path / "output1.csv" - p1 = strip_scheme(upath1) + p1 = upath1.path upath1.write_bytes(content) - assert fs is upath1.fs + assert fs._fs_token == upath1.fs._fs_token + if fs.cachable: # codespell:ignore cachable + assert fs is upath1.fs with fs.open(p1) as f: assert f.read() == content upath1.unlink() # write with fsspec, read with upath upath2 = self.path / "output2.csv" - p2 = strip_scheme(upath2) - assert fs is upath2.fs + p2 = upath2.path + if fs.cachable: # codespell:ignore cachable + assert fs is upath2.fs with fs.open(p2, "wb") as f: f.write(content) assert upath2.read_bytes() == content @@ -424,8 +422,10 @@ def test_pickling_child_path(self): assert path.storage_options == recovered_path.storage_options def test_child_path(self): - path_str = str(self.path) - path_a = UPath(path_str, "folder", **self.path.storage_options) + path_str = self.path.__vfspath__() + path_a = UPath( + path_str, "folder", protocol=self.path.protocol, **self.path.storage_options + ) path_b = self.path / "folder" assert str(path_a) == str(path_b) diff --git a/upath/tests/implementations/test_cached.py b/upath/tests/implementations/test_cached.py new file mode 100644 index 00000000..e7b757f2 --- /dev/null +++ b/upath/tests/implementations/test_cached.py @@ -0,0 +1,19 @@ +import pytest + +from upath import UPath +from upath.implementations.cached import SimpleCachePath + +from ..cases import BaseTests + + +class TestSimpleCachePath(BaseTests): + @pytest.fixture(autouse=True) + def path(self, local_testdir): + if not local_testdir.startswith("/"): + local_testdir = "/" + local_testdir + path = f"simplecache::memory:{local_testdir}" + self.path = UPath(path) + self.prepare_file_system() + + def test_is_SimpleCachePath(self): + assert isinstance(self.path, SimpleCachePath) diff --git a/upath/tests/test_chain.py b/upath/tests/test_chain.py index 54a5a696..8d91be4f 100644 --- a/upath/tests/test_chain.py +++ b/upath/tests/test_chain.py @@ -54,7 +54,6 @@ def test_chaining_upath_path(urlpath, expected): "simplecache::file:///tmp", { "target_protocol": "file", - "fo": Path("/tmp").absolute().as_posix(), "target_options": {}, }, ), @@ -130,7 +129,7 @@ def test_write_file(clear_memory_fs): ) def test_chain_parser_roundtrip(urlpath: str): parser = FSSpecChainParser() - segments = parser.unchain(urlpath, {}) + segments = parser.unchain(urlpath, protocol=None, storage_options={}) rechained, kw = parser.chain(segments) assert rechained == urlpath assert kw == {}