Skip to content
Merged
2 changes: 1 addition & 1 deletion dev/fsspec_inspector/generate_flavours.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,6 @@ def __init_subclass__(cls: Any, **kwargs):
"dir",
"blockcache",
"cached",
"simplecache",
"filecache",
]

Expand All @@ -116,6 +115,7 @@ def __init_subclass__(cls: Any, **kwargs):

FIX_METHODS = {
"GCSFileSystem": ["_strip_protocol", "_get_kwargs_from_urls", "_split_path"],
"SimpleCacheFileSystem": [],
}


Expand Down
172 changes: 116 additions & 56 deletions upath/_chain.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -153,74 +158,129 @@ 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"""

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:
1. it sets the urlpath to None for upstream filesystems that passthrough
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"""
Expand Down Expand Up @@ -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)
Expand Down
10 changes: 9 additions & 1 deletion upath/_flavour_sources.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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'
Expand Down
59 changes: 35 additions & 24 deletions upath/core.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from __future__ import annotations

import os
import posixpath
import sys
import warnings
from abc import ABCMeta
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down
Loading