diff --git a/upath/_flavour.py b/upath/_flavour.py index 336992e3..e030dff3 100644 --- a/upath/_flavour.py +++ b/upath/_flavour.py @@ -11,7 +11,6 @@ from typing import Mapping from typing import Sequence from typing import Union -from typing import cast from urllib.parse import urlsplit if sys.version_info >= (3, 12): @@ -34,8 +33,10 @@ from upath.core import UPath __all__ = [ - "FileSystemFlavourBase", + "LazyFlavourDescriptor", + "default_flavour", "upath_urijoin", + "upath_get_kwargs_from_url", ] class_registry: Mapping[str, type[AbstractFileSystem]] @@ -59,34 +60,33 @@ def wrapper(*args, **kwargs): return func -class LazyFlavourDescriptor: - """descriptor to lazily get the flavour for a given protocol""" - - def __init__(self) -> None: - self._owner = None +class AnyProtocolFileSystemFlavour(FileSystemFlavourBase): + sep: str = "/" + protocol: tuple[str, ...] = () + root_marker: str = "/" - def __set_name__(self, owner: type[UPath], name: str) -> None: - # helper to provide a more informative repr - self._owner = owner - try: - self._default_protocol = self._owner.protocols[0] - except (AttributeError, IndexError): - self._default_protocol = None + @classmethod + def _strip_protocol(cls, path: str) -> str: + protocol = get_upath_protocol(path) + if path.startswith(protocol + "://"): + path = path[len(protocol) + 3 :] + elif path.startswith(protocol + "::"): + path = path[len(protocol) + 2 :] + path = path.rstrip("/") + return path or cls.root_marker - def __get__(self, instance: UPath, owner: type[UPath]) -> WrappedFileSystemFlavour: - if instance is not None: - return WrappedFileSystemFlavour.from_protocol(instance.protocol) - elif self._default_protocol: - return WrappedFileSystemFlavour.from_protocol(self._default_protocol) - else: - return self + @staticmethod + def _get_kwargs_from_urls(path: str) -> dict[str, Any]: + return {} - def __repr__(self): - cls_name = f"{type(self).__name__}" - if self._owner is None: - return f"" + @classmethod + def _parent(cls, path): + path = cls._strip_protocol(path) + if "/" in path: + parent = path.rsplit("/", 1)[0].lstrip(cls.root_marker) + return cls.root_marker + parent else: - return f"<{cls_name} of {self._owner.__name__}>" + return cls.root_marker class WrappedFileSystemFlavour: # (pathlib_abc.FlavourBase) @@ -111,7 +111,7 @@ class WrappedFileSystemFlavour: # (pathlib_abc.FlavourBase) # indicating the following settings via the protocol. This is a # workaround to be able to implement the flavour correctly. # TODO: - # Both these settings should be configured on the UPath class?!? + # These settings should be configured on the UPath class?!? # _protocols_with_netloc_anchor = { "http", @@ -176,17 +176,7 @@ def from_protocol( UserWarning, stacklevel=2, ) - # for now let's only allow valid python identifiers as new protocol names - assert protocol.isidentifier(), f"invalid protocol: {protocol!r}" - cls_name = f"{protocol.title()}FileSystemFlavour" - # avoid circular imports - base_cls = flavour_registry["abstract"] # AbstractFileSystemFlavour - # overwrite the relevant attributes - attrs = {"protocol": protocol, "root_marker": "/"} - default = cast( - "type[FileSystemFlavourBase]", type(cls_name, (base_cls,), attrs) - ) - return cls(default) + return cls(AnyProtocolFileSystemFlavour) def __repr__(self): if isinstance(self._spec, type): @@ -378,6 +368,39 @@ def join_parsed_parts( return drv2, root2, parts2 +default_flavour = WrappedFileSystemFlavour(AnyProtocolFileSystemFlavour) + + +class LazyFlavourDescriptor: + """descriptor to lazily get the flavour for a given protocol""" + + def __init__(self) -> None: + self._owner = None + + def __set_name__(self, owner: type[UPath], name: str) -> None: + # helper to provide a more informative repr + self._owner = owner + try: + self._default_protocol = self._owner.protocols[0] + except (AttributeError, IndexError): + self._default_protocol = None + + def __get__(self, instance: UPath, owner: type[UPath]) -> WrappedFileSystemFlavour: + if instance is not None: + return WrappedFileSystemFlavour.from_protocol(instance.protocol) + elif self._default_protocol: + return WrappedFileSystemFlavour.from_protocol(self._default_protocol) + else: + return default_flavour + + def __repr__(self): + cls_name = f"{type(self).__name__}" + if self._owner is None: + return f"" + else: + return f"<{cls_name} of {self._owner.__name__}>" + + def upath_strip_protocol(pth: PathOrStr) -> str: if protocol := get_upath_protocol(pth): return WrappedFileSystemFlavour.from_protocol(protocol).strip_protocol(pth) diff --git a/upath/core.py b/upath/core.py index 5e6c156a..d68f8282 100644 --- a/upath/core.py +++ b/upath/core.py @@ -20,7 +20,6 @@ from upath._compat import str_remove_prefix from upath._compat import str_remove_suffix from upath._flavour import LazyFlavourDescriptor -from upath._flavour import WrappedFileSystemFlavour from upath._flavour import upath_get_kwargs_from_url from upath._flavour import upath_urijoin from upath._protocol import get_upath_protocol @@ -32,6 +31,8 @@ def __getattr__(name): if name == "_UriFlavour": + from upath._flavour import default_flavour + warnings.warn( "upath.core._UriFlavour should not be used anymore." " Please follow the universal_pathlib==0.2.0 migration guide at" @@ -40,7 +41,7 @@ def __getattr__(name): DeprecationWarning, stacklevel=2, ) - return WrappedFileSystemFlavour + return default_flavour elif name == "PT": warnings.warn( "upath.core.PT should not be used anymore."