"""upath.core module: UPath base class implementation""" from __future__ import annotations import sys import warnings from abc import ABCMeta from abc import abstractmethod from collections.abc import Iterator from collections.abc import Mapping from collections.abc import Sequence from copy import copy from pathlib import PurePath from types import MappingProxyType from typing import IO from typing import TYPE_CHECKING from typing import Any from typing import BinaryIO from typing import Literal from typing import NoReturn from typing import TextIO from typing import TypeVar from typing import overload from urllib.parse import SplitResult from urllib.parse import urlsplit from fsspec.registry import get_filesystem_class from fsspec.spec import AbstractFileSystem from upath._chain import DEFAULT_CHAIN_PARSER from upath._chain import Chain from upath._chain import FSSpecChainParser 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._info import UPathInfo from upath._protocol import compatible_protocol from upath._protocol import get_upath_protocol from upath._stat import UPathStatResult from upath.registry import _get_implementation_protocols from upath.registry import available_implementations from upath.registry import get_upath_class from upath.types import UNSET_DEFAULT from upath.types import JoinablePathLike from upath.types import OnNameCollisionFunc from upath.types import PathInfo from upath.types import ReadablePath from upath.types import ReadablePathLike from upath.types import StatResultType from upath.types import SupportsPathLike from upath.types import UPathParser from upath.types import WritablePath from upath.types import WritablePathLike if sys.version_info >= (3, 13): from pathlib import UnsupportedOperation else: UnsupportedOperation = NotImplementedError """Raised when an unsupported operation is called on a path object.""" if TYPE_CHECKING: import upath.implementations as _uimpl if sys.version_info >= (3, 11): from typing import Self else: from typing_extensions import Self from pydantic import GetCoreSchemaHandler from pydantic_core.core_schema import CoreSchema _MT = TypeVar("_MT") _WT = TypeVar("_WT", bound="WritablePath") __all__ = [ "UPath", "UnsupportedOperation", ] _FSSPEC_HAS_WORKING_GLOB = None def _check_fsspec_has_working_glob(): global _FSSPEC_HAS_WORKING_GLOB from fsspec.implementations.memory import MemoryFileSystem m = type("_M", (MemoryFileSystem,), {"store": {}, "pseudo_dirs": [""]})() m.touch("a.txt") m.touch("f/b.txt") g = _FSSPEC_HAS_WORKING_GLOB = len(m.glob("**/*.txt")) == 2 return g def _make_instance(cls, args, kwargs): """helper for pickling UPath instances""" # Extract _relative_base if present relative_base = kwargs.pop("_relative_base", None) instance = cls(*args, **kwargs) if relative_base is not None: instance._relative_base = relative_base return instance def _buffering2blocksize(mode: str, buffering: int) -> int | None: if not isinstance(buffering, int): raise TypeError("buffering must be an integer") if buffering == 0: # buffering disabled if "b" not in mode: # text mode raise ValueError("can't have unbuffered text I/O") return buffering elif buffering == -1: return None else: return buffering def _raise_unsupported(cls_name: str, method: str) -> NoReturn: raise UnsupportedOperation(f"{cls_name}.{method}() is unsupported") class _IncompatibleProtocolError(TypeError, ValueError): """switch to TypeError for incompatible protocols in a backward compatible way. !!! Do not use this exception directly !!! Catch TypeError instead, if you need to handle incompatible protocol errors. We'll do the switch in a future major release. """ # evil: make this look like a built-in TypeError __module__ = "builtins" __qualname__ = "TypeError" def __repr__(self) -> str: return f"TypeError({', '.join(map(repr, self.args))})" class _UPathMeta(ABCMeta): """metaclass for UPath to customize instance creation There are two main reasons for this metaclass: - support copying UPath instances via UPath(existing_upath) - force calling __init__ on instance creation for instances of a non-subclass """ if sys.version_info < (3, 11): # pathlib 3.9 and 3.10 supported `Path[str]` but # did not return a GenericAlias but the class itself? def __getitem__(cls, key): return cls def __call__(cls: type[_MT], *args: Any, **kwargs: Any) -> _MT: # create a copy if UPath class try: (arg0,) = args except ValueError: pass else: if isinstance(arg0, UPath) and not kwargs: return copy(arg0) # type: ignore[return-value] # We do this call manually, because cls could be a registered # subclass of UPath that is not directly inheriting from UPath. inst = cls.__new__(cls, *args, **kwargs) inst.__init__(*args, **kwargs) # type: ignore[misc] return inst class _UPathMixin(metaclass=_UPathMeta): """Mixin class for UPath to allow sharing some common functionality between UPath and PosixUPath/WindowsUPath. """ __slots__ = () @property @abstractmethod def parser(self) -> UPathParser: """The parser (flavour) for this UPath instance.""" raise NotImplementedError @property def _protocol(self) -> str: return self._chain.nest().protocol @_protocol.setter def _protocol(self, value: str) -> None: self._chain = self._chain.replace(protocol=value) @property def _storage_options(self) -> dict[str, Any]: return self._chain.nest().storage_options @_storage_options.setter def _storage_options(self, value: dict[str, Any]) -> None: self._chain = self._chain.replace(storage_options=value) @property @abstractmethod def _chain(self) -> Chain: raise NotImplementedError @_chain.setter @abstractmethod def _chain(self, value: Chain) -> None: raise NotImplementedError @property @abstractmethod def _chain_parser(self) -> FSSpecChainParser: raise NotImplementedError @_chain_parser.setter @abstractmethod def _chain_parser(self, value: FSSpecChainParser) -> None: raise NotImplementedError @property @abstractmethod def _fs_cached(self) -> AbstractFileSystem: raise NotImplementedError @_fs_cached.setter def _fs_cached(self, value: AbstractFileSystem): raise NotImplementedError @property @abstractmethod def _raw_urlpaths(self) -> Sequence[JoinablePathLike]: raise NotImplementedError @_raw_urlpaths.setter def _raw_urlpaths(self, value: Sequence[JoinablePathLike]) -> None: raise NotImplementedError @property @abstractmethod def _relative_base(self) -> str | None: raise NotImplementedError @_relative_base.setter def _relative_base(self, value: str | None) -> None: raise NotImplementedError # === upath.UPath PUBLIC ADDITIONAL API =========================== @property def protocol(self) -> str: """The fsspec protocol for the path. Note ---- Protocols are linked to upath and fsspec filesystems via the `upath.registry` and `fsspec.registry` modules. They basically represent the URI scheme used for the specific filesystem. Examples -------- >>> from upath import UPath >>> p0 = UPath("s3://my-bucket/path/to/file.txt") >>> p0.protocol 's3' >>> p1 = UPath("/foo/bar/baz.txt", protocol="memory") >>> p1.protocol 'memory' """ return self._protocol @property def storage_options(self) -> Mapping[str, Any]: """The read-only fsspec storage options for the path. Note ---- Storage options are specific to each fsspec filesystem and can include parameters such as authentication credentials, connection settings, and other options that affect how the filesystem interacts with the underlying storage. Examples -------- >>> from upath import UPath >>> p = UPath("s3://my-bucket/path/to/file.txt", anon=True) >>> p.storage_options['anon'] True """ return MappingProxyType(self._storage_options) @property def fs(self) -> AbstractFileSystem: """The cached fsspec filesystem instance for the path. This is the underlying fsspec filesystem instance. It's instantiated on first filesystem access and cached. Can be used to access fsspec-specific functionality not exposed by the UPath API. Examples -------- >>> from upath import UPath >>> p = UPath("s3://my-bucket/path/to/file.txt") >>> p.fs >>> p.fs.get_tags(p.path) {'VersionId': 'null', 'ContentLength': 12345, ...} """ try: return self._fs_cached except AttributeError: fs = self._fs_cached = self._fs_factory( str(self), self.protocol, self.storage_options ) return fs @property def path(self) -> str: """The path used by fsspec filesystem. FSSpec filesystems usually handle paths stripped of protocol. This property returns the path suitable for use with the underlying fsspec filesystem. It guarantees that a filesystem's strip_protocol method is applied correctly. Examples -------- >>> from upath import UPath >>> p = UPath("memory:///foo/bar.txt") >>> str(p) 'memory:///foo/bar.txt' >>> p.path '/foo/bar.txt' >>> p.fs.exists(p.path) True """ if self._relative_base is not None: try: # For relative paths, we need to resolve to absolute path current_dir = self.cwd() # type: ignore[attr-defined] except NotImplementedError: raise UnsupportedOperation( f"fsspec paths can not be relative and" f" {type(self).__name__}.cwd() is unsupported" ) from None # Join the current directory with the relative path if (self_path := str(self)) == ".": path = str(current_dir) else: path = current_dir.parser.join(str(current_dir), self_path) return self.parser.strip_protocol(path) return self._chain.active_path def joinuri(self, uri: JoinablePathLike) -> UPath: """Join with urljoin behavior for UPath instances. Examples -------- >>> from upath import UPath >>> p = UPath("https://example.com/dir/subdir/") >>> p.joinuri("file.txt") HTTPSPath('https://example.com/dir/subdir/file.txt') >>> p.joinuri("/anotherdir/otherfile.txt") HTTPSPath('https://example.com/anotherdir/otherfile.txt') >>> p.joinuri("memory:///foo/bar.txt" MemoryPath('memory:///foo/bar.txt') """ # short circuit if the new uri uses a different protocol other_protocol = get_upath_protocol(uri) if other_protocol and other_protocol != self._protocol: return UPath(uri) return UPath( upath_urijoin(str(self), str(uri)), protocol=other_protocol or self._protocol, **self.storage_options, ) # === upath.UPath CUSTOMIZABLE API ================================ @classmethod def _transform_init_args( cls, args: tuple[JoinablePathLike, ...], protocol: str, storage_options: dict[str, Any], ) -> tuple[tuple[JoinablePathLike, ...], str, dict[str, Any]]: """allow customization of init args in subclasses""" return args, protocol, storage_options @classmethod def _parse_storage_options( cls, urlpath: str, protocol: str, storage_options: Mapping[str, Any], ) -> dict[str, Any]: """Parse storage_options from the urlpath""" pth_storage_options = upath_get_kwargs_from_url(urlpath) return {**pth_storage_options, **storage_options} @classmethod def _fs_factory( cls, urlpath: str, protocol: str, storage_options: Mapping[str, Any], ) -> AbstractFileSystem: """Instantiate the filesystem_spec filesystem class""" fs_cls = get_filesystem_class(protocol) return fs_cls(**storage_options) # === upath.UPath constructor ===================================== _protocol_dispatch: bool | None = None def __new__( # noqa C901 cls, *args: JoinablePathLike, protocol: str | None = None, chain_parser: FSSpecChainParser = DEFAULT_CHAIN_PARSER, **storage_options: Any, ) -> UPath: # narrow type if not issubclass(cls, UPath): raise TypeError("UPath.__new__ can't instantiate non-UPath classes") # deprecate 'scheme' if "scheme" in storage_options: warnings.warn( "use 'protocol' kwarg instead of 'scheme'", DeprecationWarning, stacklevel=2, ) protocol = storage_options.pop("scheme") # determine the protocol try: pth_protocol = get_upath_protocol( args[0] if args else "", protocol=protocol, storage_options=storage_options, ) except ValueError as e: if "incompatible with" in str(e): raise _IncompatibleProtocolError(str(e)) from e raise # subclasses should default to their own protocol if protocol is None and cls is not UPath: impl_protocols = _get_implementation_protocols(cls) if not pth_protocol and impl_protocols: pth_protocol = impl_protocols[0] elif pth_protocol and pth_protocol not in impl_protocols: msg_protocol = pth_protocol if not pth_protocol: msg_protocol = "'' (empty string)" msg = ( f"{cls.__name__!s}(...) detected protocol {msg_protocol!s}" f" which is incompatible with {cls.__name__}." ) if not pth_protocol or pth_protocol not in available_implementations(): msg += ( " Did you forget to register the subclass for this protocol" " with upath.registry.register_implementation()?" ) raise _IncompatibleProtocolError(msg) # determine which UPath subclass to dispatch to upath_cls: type[UPath] | None if cls._protocol_dispatch or cls._protocol_dispatch is None: upath_cls = get_upath_class(protocol=pth_protocol) if upath_cls is None: raise ValueError(f"Unsupported filesystem: {pth_protocol!r}") else: # user subclasses can request to disable protocol dispatch # by setting MyUPathSubclass._protocol_dispatch to `False`. # This will effectively ignore the registered UPath # implementations and return an instance of MyUPathSubclass. # This be useful if a subclass wants to extend the UPath # api, and it is fine to rely on the default implementation # for all supported user protocols. # # THIS IS DEPRECATED! # Use upath.extensions.ProxyUPath to extend the UPath API warnings.warn( f"{cls.__name__}._protocol_dispatch = False is deprecated and" " will be removed in future universal_pathlib versions." " To extend the UPath API, subclass upath.extensions.ProxyUPath", DeprecationWarning, stacklevel=2, ) upath_cls = cls if issubclass(upath_cls, cls): pass elif not issubclass(upath_cls, UPath): raise RuntimeError("UPath.__new__ expected cls to be subclass of UPath") else: msg_protocol = pth_protocol if not pth_protocol: msg_protocol = "'' (empty string)" msg = ( f"{cls.__name__!s}(...) detected protocol {msg_protocol!s}" f" which is incompatible with {cls.__name__}." ) if ( # find a better way (not pth_protocol and cls.__name__ not in ["CloudPath", "LocalPath"]) or pth_protocol and pth_protocol not in available_implementations() ): msg += ( " Did you forget to register the subclass for this protocol" " with upath.registry.register_implementation()?" ) raise _IncompatibleProtocolError(msg) return object.__new__(upath_cls) def __init__( self, *args: JoinablePathLike, protocol: str | None = None, chain_parser: FSSpecChainParser = DEFAULT_CHAIN_PARSER, **storage_options: Any, ) -> None: """Initialize a UPath instance When instantiating a `UPath`, the detected or provided protocol determines the `UPath` subclass that will be instantiated. The protocol is looked up via the `get_upath_protocol` function, which loads the registered `UPath` implementation from the registry. If no `UPath` implementation is found for the detected protocol, but a registered `fsspec` filesystem exists for the protocol, a default dynamically created `UPath` implementation will be used. Parameters ---------- *args : The path (or uri) segments to construct the UPath from. The first argument is used to detect the protocol if no protocol is provided. protocol : The protocol to use for the path. chain_parser : A chain parser instance for chained urlpaths. _(experimental)_ **storage_options : Additional storage options for the path. """ # todo: avoid duplicating this call from __new__ protocol = get_upath_protocol( args[0] if args else "", protocol=protocol, storage_options=storage_options, ) args, protocol, storage_options = type(self)._transform_init_args( args, protocol, storage_options ) # check that UPath subclasses in args are compatible # TODO: # Future versions of UPath could verify that storage_options # can be combined between UPath instances. Not sure if this # is really necessary though. A warning might be enough... if not compatible_protocol(protocol, *args): raise ValueError("can't combine incompatible UPath protocols") # subclasses should default to their own protocol if not protocol: impl_protocols = _get_implementation_protocols(type(self)) if impl_protocols: protocol = impl_protocols[0] if args: args0 = args[0] if isinstance(args0, UPath): storage_options = { **args0._chain.nest().storage_options, **storage_options, } str_args0 = args0.__vfspath__() else: if hasattr(args0, "__fspath__") and args0.__fspath__ is not None: str_args0 = args0.__fspath__() elif hasattr(args0, "__vfspath__") and args0.__vfspath__ is not None: str_args0 = args0.__vfspath__() elif isinstance(args0, str): str_args0 = args0 else: raise TypeError( "argument should be a UPath, str, " f"or support __vfspath__ or __fspath__, not {type(args0)!r}" ) storage_options = type(self)._parse_storage_options( str_args0, protocol, storage_options ) else: str_args0 = "." segments = chain_parser.unchain( str_args0, protocol=protocol, storage_options=storage_options, ) # FIXME: normalization needs to happen in unchain already... chain = Chain.from_list(Chain.from_list(segments).to_list()) if len(args) > 1: flavour = WrappedFileSystemFlavour.from_protocol(chain.active_path_protocol) joined = flavour.join(chain.active_path, *args[1:]) stripped = flavour.strip_protocol(joined) chain = chain.replace(path=stripped) self._chain = chain self._chain_parser = chain_parser self._raw_urlpaths = args self._relative_base = None # --- deprecated attributes --------------------------------------- @property def _url(self) -> SplitResult: # TODO: # _url should be deprecated, but for now there is no good way of # accessing query parameters from urlpaths... return urlsplit(self.__str__()) class UPath(_UPathMixin, WritablePath, ReadablePath): """Base class for pathlike paths backed by an fsspec filesystem. Note ---- The following attributes and methods are specific to UPath instances and are not available on pathlib.Path instances. Attributes ---------- protocol : The fsspec protocol for the path. storage_options : The fsspec storage options for the path. path : The path that a fsspec filesystem can use. fs : The cached fsspec filesystem instance for the path. Methods ------- joinuri(*parts) : Join URI parts to this path. Info ---- Below are pathlib attributes and methods available on UPath instances. Attributes ---------- drive : The drive component of the path. root : The root component of the path. anchor : The concatenation of the drive and root. parent : The logical parent of the path. parents : An immutable sequence providing access to the logical ancestors of the path. name : The final path component, excluding the drive and root, if any. suffix : The file extension of the final component, if any. suffixes : A list of the path's file extensions. stem : The final path component, without its suffix. info : Filesystem information about the path. parser : The path parser instance for parsing path segments. Methods ------- __truediv__(key) : Combine this path with the argument using the `/` operator. __rtruediv__(key) : Combine the argument with this path using the `/` operator. as_posix() : Return the string representation of the path with forward slashes. is_absolute() : Return True if the path is absolute. is_relative_to(other) : Return True if the path is relative to another path. is_reserved() : Return True if the path is reserved under Windows. joinpath(*pathsegments) : Combine this path with one or several arguments, and return a new path. full_match(pattern, *, case_sensitive=None) : Match this path against the provided glob-style pattern. match(pattern, *, case_sensitive=None) : Match this path against the provided glob-style pattern. relative_to(other, walk_up=False) : Return a version of this path relative to another path. with_name(name) : Return a new path with the name changed. with_stem(stem) : Return a new path with the stem changed. with_suffix(suffix) : Return a new path with the suffix changed. with_segments(*pathsegments) : Construct a new path object from any number of path-like objects. from_uri(uri) : Return a new path from the given URI. as_uri() : Return the path as a URI. home() : Return a new path pointing to the user's home directory. expanduser() : Return a new path with expanded `~` constructs. cwd() : Return a new path pointing to the current working directory. absolute() : Make the path absolute, without normalization or resolving symlinks. resolve(strict=False) : Make the path absolute, resolving any symlinks. readlink() : Return the path to which the symbolic link points. stat(*, follow_symlinks=True) : Return the result of the stat() system call on this path. lstat() : Like stat(), but if the path points to a symlink, return the symlink's information. exists(*, follow_symlinks=True) : Return True if the path exists. is_file(*, follow_symlinks=True) : Return True if the path is a regular file. is_dir(*, follow_symlinks=True) : Return True if the path is a directory. is_symlink() : Return True if the path is a symbolic link. is_junction() : Return True if the path is a junction. is_mount() : Return True if the path is a mount point. is_socket() : Return True if the path is a socket. is_fifo() : Return True if the path is a FIFO. is_block_device() : Return True if the path is a block device. is_char_device() : Return True if the path is a character device. samefile(other_path) : Return True if this path points to the same file as other_path. open(mode='r', buffering=-1, encoding=None, errors=None, newline=None) : Open the file pointed to by the path. read_text(encoding=None, errors=None, newline=None) : Open the file in text mode, read it, and close the file. read_bytes() : Open the file in bytes mode, read it, and close the file. write_text(data, encoding=None, errors=None, newline=None) : Open the file in text mode, write to it, and close the file. write_bytes(data) : Open the file in bytes mode, write to it, and close the file. iterdir() : Yield path objects of the directory contents. glob(pattern, *, case_sensitive=None) : Iterate over this subtree and yield all existing files matching the given pattern. rglob(pattern, *, case_sensitive=None) : Recursively yield all existing files matching the given pattern. walk(top_down=True, on_error=None, follow_symlinks=False) : Generate the file names in a directory tree by walking the tree. touch(mode=0o666, exist_ok=True) : Create this file with the given access mode, if it doesn't exist. mkdir(mode=0o777, parents=False, exist_ok=False) : Create a new directory at this given path. symlink_to(target, target_is_directory=False) : Make this path a symbolic link pointing to target. hardlink_to(target) : Make this path a hard link pointing to the same file as target. copy(target, *, follow_symlinks=True, preserve_metadata=False) : Copy the contents of this file to the target file. copy_into(target_dir, *, follow_symlinks=True, preserve_metadata=False) : Copy this file or directory into the target directory. rename(target) : Rename this path to the target path. replace(target) : Rename this path to the target path, overwriting if that path exists. move(target) : Move this file or directory tree to the target path. move_into(target_dir) : Move this file or directory into the target directory. unlink(missing_ok=False) : Remove this file or link. rmdir() : Remove this directory. owner(*, follow_symlinks=True) : Return the login name of the file owner. group(*, follow_symlinks=True) : Return the group name of the file gid. chmod(mode, *, follow_symlinks=True) : Change the permissions of the path. lchmod(mode) : Like chmod() but, if the path points to a symlink, modify the symlink's permissions. """ __slots__ = ( "_chain", "_chain_parser", "_fs_cached", "_raw_urlpaths", "_relative_base", ) if TYPE_CHECKING: # noqa: C901 _chain: Chain _chain_parser: FSSpecChainParser _fs_cached: AbstractFileSystem _raw_urlpaths: Sequence[JoinablePathLike] _relative_base: str | None @overload def __new__( cls, ) -> Self: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["simplecache"], **_: Any, ) -> _uimpl.cached.SimpleCachePath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["gcs", "gs"], **_: Any, ) -> _uimpl.cloud.GCSPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["s3", "s3a"], **_: Any, ) -> _uimpl.cloud.S3Path: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["az", "abfs", "abfss", "adl"], **_: Any, ) -> _uimpl.cloud.AzurePath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["hf"], **_: Any, ) -> _uimpl.cloud.HfPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["data"], **_: Any, ) -> _uimpl.data.DataPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["ftp"], **_: Any, ) -> _uimpl.ftp.FTPPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["github"], **_: Any, ) -> _uimpl.github.GitHubPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["hdfs"], **_: Any, ) -> _uimpl.hdfs.HDFSPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["http", "https"], **_: Any, ) -> _uimpl.http.HTTPPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["file", "local"], **_: Any, ) -> _uimpl.local.FilePath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["memory"], **_: Any, ) -> _uimpl.memory.MemoryPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["sftp", "ssh"], **_: Any, ) -> _uimpl.sftp.SFTPPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["smb"], **_: Any, ) -> _uimpl.smb.SMBPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["tar"], **_: Any, ) -> _uimpl.tar.TarPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["webdav"], **_: Any, ) -> _uimpl.webdav.WebdavPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal["zip"], **_: Any, ) -> _uimpl.zip.ZipPath: ... if sys.platform == "win32": @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal[""], **_: Any, ) -> _uimpl.local.WindowsUPath: ... else: @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: Literal[""], **_: Any, ) -> _uimpl.local.PosixUPath: ... @overload # noqa: E301 def __new__( cls, *args: JoinablePathLike, protocol: str | None = ..., **_: Any, ) -> Self: ... def __new__( cls, *args: JoinablePathLike, protocol: str | None = ..., chain_parser: FSSpecChainParser = ..., **storage_options: Any, ) -> Self: ... # === JoinablePath attributes ===================================== parser: UPathParser = LazyFlavourDescriptor() # type: ignore[assignment] def with_segments(self, *pathsegments: JoinablePathLike) -> Self: """Construct a new path object from any number of path-like objects.""" # we change joinpath behavior if called from a relative path # this is not fully ideal, but currently the best way to move forward if is_relative := self._relative_base is not None: pathsegments = (self._relative_base, *pathsegments) new_instance = type(self)( *pathsegments, protocol=self._protocol, **self.storage_options, ) if hasattr(self, "_fs_cached"): new_instance._fs_cached = self._fs_cached if is_relative: new_instance._relative_base = self._relative_base return new_instance def __str__(self) -> str: if self._relative_base is not None: active_path = self._chain.active_path stripped_base = self.parser.strip_protocol( self._relative_base ).removesuffix(self.parser.sep) if not active_path.startswith(stripped_base): raise RuntimeError( f"{active_path!r} is not a subpath of {stripped_base!r}" ) return ( active_path.removeprefix(stripped_base).removeprefix(self.parser.sep) or "." ) else: return self._chain_parser.chain(self._chain.to_list())[0] def __vfspath__(self) -> str: if self._relative_base is not None: return self.__str__() else: return self.path def __repr__(self) -> str: cls_name = type(self).__name__ path = self.__vfspath__() if self._relative_base is not None: return f"" else: return f"{cls_name}({path!r}, protocol={self._protocol!r})" # === JoinablePath overrides ====================================== @property def parts(self) -> Sequence[str]: """Provides sequence-like access to the filesystem path components. Examples -------- >>> from upath import UPath >>> p = UPath("s3://my-bucket/path/to/file.txt") >>> p.parts ('my-bucket/', 'path', 'to', 'file.txt') >>> p2 = UPath("/foo/bar/baz.txt", protocol="memory") >>> p2.parts ('/', 'foo', 'bar', 'baz.txt') """ # For relative paths, return parts of the relative path only if self._relative_base is not None: rel_str = str(self) if rel_str == ".": return () return tuple(rel_str.split(self.parser.sep)) split = self.parser.split sep = self.parser.sep path = self._chain.active_path drive = self.parser.splitdrive(self._chain.active_path)[0] stripped_path = self.parser.strip_protocol(path) if stripped_path: _, _, tail = path.partition(stripped_path) path = stripped_path + tail parent, name = split(path) names = [] while path != parent: names.append(name) path = parent parent, name = split(path) if names and names[-1] == drive: names = names[:-1] if names and names[-1].startswith(sep): parts = [*names[:-1], names[-1].removeprefix(sep), drive + sep] else: parts = [*names, drive + sep] return tuple(reversed(parts)) def with_name(self, name: str) -> Self: """Return a new path with the file name changed.""" split = self.parser.split if self.parser.sep in name: # `split(name)[0]` raise ValueError(f"Invalid name {name!r}") _path = self.__vfspath__() _path = _path.removesuffix(split(_path)[1]) + name return self.with_segments(_path) @property def anchor(self) -> str: """The concatenation of the drive and root or an empty string.""" if self._relative_base is not None: return "" return self.drive + self.root @property def parent(self) -> Self: """The logical parent of the path. Examples -------- >>> from upath import UPath >>> p = UPath("s3://my-bucket/path/to/file.txt") >>> p.parent S3Path('s3://my-bucket/path/to') """ if self._relative_base is not None: if str(self) == ".": return self else: # this needs to be revisited... pth = type(self)( self._relative_base, str(self), protocol=self._protocol, **self.storage_options, ) parent = pth.parent parent._relative_base = self._relative_base return parent return super().parent @property def parents(self) -> Sequence[Self]: """A sequence providing access to the logical ancestors of the path. Examples -------- >>> from upath import UPath >>> p = UPath("memory:///foo/bar/baz.txt") >>> list(p.parents) [ MemoryPath('memory:///foo/bar'), MemoryPath('memory:///foo'), MemoryPath('memory:///'), ] """ if self._relative_base is not None: parents = [] parent = self while True: if str(parent) == ".": break parent = parent.parent parents.append(parent) return tuple(parents) return super().parents def joinpath(self, *pathsegments: JoinablePathLike) -> Self: """Combine this path with one or several arguments, and return a new path. For one argument, this is equivalent to using the `/` operator. Examples -------- >>> from upath import UPath >>> p = UPath("s3://my-bucket/path/to") >>> p.joinpath("file.txt") S3Path('s3://my-bucket/path/to/file.txt') """ return self.with_segments(self.__vfspath__(), *pathsegments) def __truediv__(self, key: JoinablePathLike) -> Self: try: return self.with_segments(self.__vfspath__(), key) except TypeError: return NotImplemented def __rtruediv__(self, key: JoinablePathLike) -> Self: try: return self.with_segments(key, self.__vfspath__()) except TypeError: return NotImplemented # === ReadablePath attributes ===================================== @property def info(self) -> PathInfo: """ A PathInfo object that exposes the file type and other file attributes of this path. Returns ------- : UPathInfo The UPathInfo object for this path. """ return UPathInfo(self) def iterdir(self) -> Iterator[Self]: """Yield path objects of the directory contents. Examples -------- >>> from upath import UPath >>> p = UPath("memory:///foo/") >>> p.joinpath("bar.txt").touch() >>> p.joinpath("baz.txt").touch() >>> for child in p.iterdir(): ... print(child) MemoryPath('memory:///foo/bar.txt') MemoryPath('memory:///foo/baz.txt') """ sep = self.parser.sep base = self if self.parts[-1:] == ("",): base = self.parent fs = base.fs base_path = base.path if not fs.isdir(base_path): raise NotADirectoryError(str(self)) for name in fs.listdir(base_path): # fsspec returns dictionaries if isinstance(name, dict): name = name.get("name") if name in {".", ".."}: # Yielding a path object for these makes little sense continue # only want the path name with iterdir _, _, name = name.removesuffix(sep).rpartition(self.parser.sep) yield base.with_segments(base_path, name) 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") @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 | UPath: """ Recursively copy this file or directory tree to the given destination. """ if isinstance(target, str): proto = get_upath_protocol(target) if proto != self.protocol: target = UPath(target) else: target = self.with_segments(target) elif not isinstance(target, UPath): target = UPath(target) if target.is_dir(): raise IsADirectoryError(str(target)) return super().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 | UPath: """ Copy this file or directory tree into the given existing directory. """ if isinstance(target_dir, str): proto = get_upath_protocol(target_dir) if proto != self.protocol: target_dir = UPath(target_dir) else: target_dir = self.with_segments(target_dir) elif not isinstance(target_dir, UPath): target_dir = UPath(target_dir) if not target_dir.exists(): raise FileNotFoundError(str(target_dir)) if not target_dir.is_dir(): raise NotADirectoryError(str(target_dir)) return super().copy_into(target_dir, **kwargs) @overload def move(self, target: _WT, **kwargs: Any) -> _WT: ... @overload def move(self, target: SupportsPathLike | str, **kwargs: Any) -> Self: ... def move(self, target: _WT | SupportsPathLike | str, **kwargs: Any) -> _WT | UPath: """ Recursively move this file or directory tree to the given destination. """ target = self.copy(target, **kwargs) self.fs.rm(self.path, recursive=self.is_dir()) return target @overload def move_into(self, target_dir: _WT, **kwargs: Any) -> _WT: ... @overload def move_into(self, target_dir: SupportsPathLike | str, **kwargs: Any) -> Self: ... def move_into( self, target_dir: _WT | SupportsPathLike | str, **kwargs: Any ) -> _WT | UPath: """ Move this file or directory tree into the given existing directory. """ name = self.name if not name: raise ValueError(f"{self!r} has an empty name") elif hasattr(target_dir, "with_segments"): target = target_dir.with_segments(target_dir, name) # type: ignore elif isinstance(target_dir, PurePath): target = UPath(target_dir, name) else: target = self.with_segments(target_dir, name) td = target.parent if not td.exists(): raise FileNotFoundError(str(td)) elif not td.is_dir(): raise NotADirectoryError(str(td)) return self.move(target) def _copy_from( self, source: ReadablePath, follow_symlinks: bool = True, on_name_collision: OnNameCollisionFunc | None = None, **kwargs: Any, ) -> None: """ UPath custom:: Recursively copy the given path to this path. """ # fixme: it would be best if this would be upstreamed from pathlib_abc import vfsopen from pathlib_abc import vfspath from pathlib_abc._os import copyfileobj from pathlib_abc._os import ensure_different_files stack: list[tuple[ReadablePath, WritablePath]] = [(source, self)] while stack: src, dst = stack.pop() info = src.info if not follow_symlinks and info.is_symlink(): dst.symlink_to(vfspath(src.readlink()), src.info.is_dir()) elif on_name_collision and info.is_file() and info.is_dir(): dst_file, dst_dir = on_name_collision(src, dst) if dst_file is not None: ensure_different_files(src, dst_file) with vfsopen(src, "rb") as source_f: with vfsopen(dst_file, "wb") as target_f: copyfileobj(source_f, target_f) if dst_dir is not None: children = src.iterdir() dst_dir.mkdir() # feed through dict.fromkeys to remove duplicates for child in dict.fromkeys(children): stack.append((child, dst_dir.joinpath(child.name))) elif info.is_dir(): children = src.iterdir() dst.mkdir() # feed through dict.fromkeys to remove duplicates for child in dict.fromkeys(children): stack.append((child, dst.joinpath(child.name))) else: ensure_different_files(src, dst) with vfsopen(src, "rb") as source_f: with vfsopen(dst, "wb") as target_f: copyfileobj(source_f, target_f) # --- WritablePath attributes ------------------------------------- def symlink_to( self, target: ReadablePathLike, target_is_directory: bool = False, ) -> None: _raise_unsupported(type(self).__name__, "symlink_to") def mkdir( self, mode: int = 0o777, parents: bool = False, exist_ok: bool = False, ) -> None: """ Create a new directory at this given path. """ if parents and not exist_ok and self.exists(): raise FileExistsError(str(self)) try: self.fs.mkdir( self.path, create_parents=parents, mode=mode, ) except FileExistsError: if not exist_ok: raise FileExistsError(str(self)) if not self.is_dir(): raise FileExistsError(str(self)) def __open_writer__(self, mode: Literal["a", "w", "x"]) -> BinaryIO: return self.fs.open(self.path, mode=f"{mode}b") # --- upath overrides --------------------------------------------- @overload def open( self, mode: Literal["r", "w", "a"] = ..., buffering: int = ..., encoding: str = ..., errors: str = ..., newline: str = ..., **fsspec_kwargs: Any, ) -> TextIO: ... @overload def open( self, mode: Literal["rb", "wb", "ab"] = ..., 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]: """ Open the file pointed by this path and return a file object, as the built-in open() function does. Parameters ---------- mode: Opening mode. Default is 'r'. buffering: Default is the block size of the underlying fsspec filesystem. encoding: Encoding is only used in text mode. Default is None. errors: Error handling for encoding. Only used in text mode. Default is None. newline: Newline handling. Only used in text mode. Default is None. **fsspec_kwargs: Additional options for the fsspec filesystem. """ # match the signature of pathlib.Path.open() if buffering is not UNSET_DEFAULT: if "block_size" in fsspec_kwargs: raise TypeError("cannot specify both 'buffering' and 'block_size'") block_size = _buffering2blocksize(mode, buffering) if block_size is not None: fsspec_kwargs.setdefault("block_size", block_size) if encoding is not UNSET_DEFAULT: fsspec_kwargs["encoding"] = encoding if errors is not UNSET_DEFAULT: fsspec_kwargs["errors"] = errors if newline is not UNSET_DEFAULT: fsspec_kwargs["newline"] = newline return self.fs.open(self.path, mode=mode, **fsspec_kwargs) # === pathlib.Path ================================================ def stat( self, *, follow_symlinks: bool = True, ) -> StatResultType: """ Return the result of the stat() system call on this path, like os.stat() does. Info ---- For fsspec filesystems follow_symlinks is currently ignored. Returns ------- : UPathStatResult The upath stat result for this path, emulating `os.stat_result`. """ if not follow_symlinks: warnings.warn( f"{type(self).__name__}.stat(follow_symlinks=False):" " is currently ignored.", UserWarning, stacklevel=2, ) return UPathStatResult.from_info(self.fs.info(self.path)) def lstat(self) -> StatResultType: return self.stat(follow_symlinks=False) def chmod(self, mode: int, *, follow_symlinks: bool = True) -> None: _raise_unsupported(type(self).__name__, "chmod") def exists(self, *, follow_symlinks: bool = True) -> bool: """ Whether this path exists. Info ---- For fsspec filesystems follow_symlinks is currently ignored. """ if not follow_symlinks: warnings.warn( f"{type(self).__name__}.exists() follow_symlinks=False" " is currently ignored.", UserWarning, stacklevel=2, ) return self.fs.exists(self.path) def is_dir(self, *, follow_symlinks: bool = True) -> bool: """ Whether this path is a directory. """ if not follow_symlinks: warnings.warn( f"{type(self).__name__}.is_dir() follow_symlinks=False" " is currently ignored.", UserWarning, stacklevel=2, ) return self.fs.isdir(self.path) def is_file(self, *, follow_symlinks: bool = True) -> bool: """ Whether this path is a regular file. """ if not follow_symlinks: warnings.warn( f"{type(self).__name__}.is_file() follow_symlinks=False" " is currently ignored.", UserWarning, stacklevel=2, ) return self.fs.isfile(self.path) def is_mount(self) -> bool: """ Check if this path is a mount point Info ---- For fsspec filesystems this is always False. """ return False def is_symlink(self) -> bool: """ Whether this path is a symbolic link. """ try: info = self.fs.info(self.path) if "islink" in info: return bool(info["islink"]) except FileNotFoundError: return False return False def is_junction(self) -> bool: """ Whether this path is a junction. Info ---- For fsspec filesystems this is always False. """ return False def is_block_device(self) -> bool: """ Whether this path is a block device. Info ---- For fsspec filesystems this is always False. """ return False def is_char_device(self) -> bool: """ Whether this path is a character device. Info ---- For fsspec filesystems this is always False. """ return False def is_fifo(self) -> bool: """ Whether this path is a FIFO (named pipe). Info ---- For fsspec filesystems this is always False. """ return False def is_socket(self) -> bool: """ Whether this path is a socket. Info ---- For fsspec filesystems this is always False. """ return False def is_reserved(self) -> bool: """ Whether this path is reserved under Windows. Info ---- For fsspec filesystems this is always False. """ return False def expanduser(self) -> Self: """Return a new path with expanded `~` constructs. Info ---- For fsspec filesystems this is currently a no-op. """ return self def glob( self, pattern: str, *, case_sensitive: bool | None = None, recurse_symlinks: bool = False, ) -> Iterator[Self]: """Iterate over this subtree and yield all existing files (of any kind, including directories) matching the given relative pattern.""" if case_sensitive is not None: warnings.warn( "UPath.glob(): case_sensitive is currently ignored.", UserWarning, stacklevel=2, ) if recurse_symlinks: warnings.warn( "UPath.glob(): recurse_symlinks=True is currently ignored.", UserWarning, stacklevel=2, ) if self._relative_base is not None: self = self.absolute() path_pattern = self.joinpath(pattern).path sep = self.parser.sep base = self.path for name in self.fs.glob(path_pattern): name = name.removeprefix(base).removeprefix(sep) yield self.joinpath(name) def rglob( self, pattern: str, *, case_sensitive: bool | None = None, recurse_symlinks: bool = False, ) -> Iterator[Self]: """Recursively yield all existing files (of any kind, including directories) matching the given relative pattern, anywhere in this subtree. """ if case_sensitive is not None: warnings.warn( "UPath.glob(): case_sensitive is currently ignored.", UserWarning, stacklevel=2, ) if recurse_symlinks: warnings.warn( "UPath.glob(): recurse_symlinks=True is currently ignored.", UserWarning, stacklevel=2, ) if _FSSPEC_HAS_WORKING_GLOB is None: _check_fsspec_has_working_glob() if _FSSPEC_HAS_WORKING_GLOB: r_path_pattern = self.joinpath("**", pattern).path sep = self.parser.sep base = self.path for name in self.fs.glob(r_path_pattern): name = name.removeprefix(base).removeprefix(sep) yield self.joinpath(name) else: path_pattern = self.joinpath(pattern).path r_path_pattern = self.joinpath("**", pattern).path sep = self.parser.sep base = self.path seen = set() for p in (path_pattern, r_path_pattern): for name in self.fs.glob(p): name = name.removeprefix(base).removeprefix(sep) if name in seen: continue else: seen.add(name) yield self.joinpath(name) def owner(self, *, follow_symlinks: bool = True) -> str: _raise_unsupported(type(self).__name__, "owner") def group(self, *, follow_symlinks: bool = True) -> str: _raise_unsupported(type(self).__name__, "group") def absolute(self) -> Self: """Return an absolute version of this path No normalization or symlink resolution is performed. Use resolve() to resolve symlinks and remove '..' segments. """ if self._relative_base is not None: return self.cwd().joinpath(self.__vfspath__()) return self def is_absolute(self) -> bool: """True if the path is absolute (has both a root and, if applicable, a drive).""" if self._relative_base is not None: return False else: return self.parser.isabs(self.__vfspath__()) def __eq__(self, other: object) -> bool: """UPaths are considered equal if their protocol, path and storage_options are equal.""" if not isinstance(other, UPath): return NotImplemented # For relative paths, compare the string representation instead of path if ( self._relative_base is not None or getattr(other, "_relative_base", None) is not None ): # If both are relative paths, compare just the relative strings if ( self._relative_base is not None and getattr(other, "_relative_base", None) is not None ): return str(self) == str(other) else: # One is relative, one is not - they can't be equal return False return ( self.__vfspath__() == other.__vfspath__() and self.protocol == other.protocol and self.storage_options == other.storage_options ) def __hash__(self) -> int: """The returned hash is based on the protocol and path only. Note: in the future, if hash collisions become an issue, we can add `fsspec.utils.tokenize(storage_options)` """ return hash((self.protocol, self.__vfspath__())) def __lt__(self, other: object) -> bool: if not isinstance(other, UPath) or self.parser is not other.parser: return NotImplemented return self.__vfspath__() < other.__vfspath__() def __le__(self, other: object) -> bool: if not isinstance(other, UPath) or self.parser is not other.parser: return NotImplemented return self.__vfspath__() <= other.__vfspath__() def __gt__(self, other: object) -> bool: if not isinstance(other, UPath) or self.parser is not other.parser: return NotImplemented return self.__vfspath__() > other.__vfspath__() def __ge__(self, other: object) -> bool: if not isinstance(other, UPath) or self.parser is not other.parser: return NotImplemented return self.__vfspath__() >= other.__vfspath__() def resolve(self, strict: bool = False) -> Self: """ Make the path absolute, resolving all symlinks on the way and also normalizing it. """ if self._relative_base is not None: self = self.absolute() _parts = self.parts # Do not attempt to normalize path if no parts are dots if ".." not in _parts and "." not in _parts: return self resolved: list[str] = [] resolvable_parts = _parts[1:] for part in resolvable_parts: if part == "..": if resolved: resolved.pop() elif part != ".": resolved.append(part) return self.with_segments(*_parts[:1], *resolved) def touch(self, mode: int = 0o666, exist_ok: bool = True) -> None: """Create this file with the given access mode, if it doesn't exist.""" exists = self.fs.exists(self.path) if exists and not exist_ok: raise FileExistsError(str(self)) if not exists: try: self.fs.touch(self.path, truncate=True) except NotImplementedError: _raise_unsupported(type(self).__name__, "touch") else: try: self.fs.touch(self.path, truncate=False) except (NotImplementedError, ValueError): pass # unsupported by filesystem def lchmod(self, mode: int) -> None: _raise_unsupported(type(self).__name__, "lchmod") def unlink(self, missing_ok: bool = False) -> None: """ Remove this file or link. If the path is a directory, use rmdir() instead. """ if not self.exists(): if not missing_ok: raise FileNotFoundError(str(self)) return self.fs.rm(self.path, recursive=False) def rmdir(self, recursive: bool = True) -> None: # fixme: non-standard """ Remove this directory. Warning ------- This method is non-standard compared to pathlib.Path.rmdir(), as it supports a `recursive` parameter to remove non-empty directories and defaults to recursive deletion. This behavior is likely to change in future releases once `.delete()` is introduced. """ if not self.is_dir(): raise NotADirectoryError(str(self)) if not recursive and next(self.iterdir()): # type: ignore[arg-type] raise OSError(f"Not recursive and directory not empty: {self}") self.fs.rm(self.path, recursive=recursive) def rename( self, target: WritablePathLike, *, # note: non-standard compared to pathlib recursive: bool = UNSET_DEFAULT, maxdepth: int | None = UNSET_DEFAULT, **kwargs: Any, ) -> Self: """ Rename this file or directory to the given target. The target path may be absolute or relative. Relative paths are interpreted relative to the current working directory, *not* the directory of the Path object. Returns the new Path instance pointing to the target path. Info ---- For filesystems that don't have a root character, i.e. for which relative paths can be ambiguous, you can explicitly indicate a relative path via prefixing with `./` Warning ------- This method is non-standard compared to pathlib.Path.rename(), as it supports `recursive` and `maxdepth` parameters for directory moves. This will be revisited in future releases. It's better to use `.move()` or `.move_into()` to avoid running into future compatibility issues. """ # check protocol compatibility 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}" ) # ensure target is an absolute UPath if not isinstance(target, type(self)): if isinstance(target, (UPath, PurePath)): target_str = target.as_posix() else: target_str = str(target) if target_protocol: # target protocol provided indicates absolute path target = self.with_segments(target_str) elif self.anchor and target_str.startswith(self.anchor): # self.anchor can be used to indicate absolute path target = self.with_segments(target_str) elif not self.anchor and target_str.startswith("./"): # indicate relative via "./" target = ( self.cwd() .joinpath(target_str.removeprefix("./")) .relative_to(self.cwd()) ) else: # all other cases target = self.cwd().joinpath(target_str).relative_to(self.cwd()) # return early if renaming to same path if target == self: return self # ensure source and target are absolute source_abs = self.absolute() target_abs = target.absolute() # avoid calling .resolve for if not needed if ".." in target_abs.parts or "." in target_abs.parts: target_abs = target_abs.resolve() if kwargs: warnings.warn( "Passing additional keyword arguments to " f"{type(self).__name__}.rename() is deprecated and will be" " removed in future univeral-pathlib versions.", DeprecationWarning, stacklevel=2, ) if recursive is not UNSET_DEFAULT: warnings.warn( f"{type(self).__name__}.rename()'s `recursive` keyword argument is" " deprecated and will be removed in future universal-pathlib versions." f" Please use {type(self).__name__}.move() or .move_into() instead.", DeprecationWarning, stacklevel=2, ) kwargs["recursive"] = recursive if maxdepth is not UNSET_DEFAULT: warnings.warn( f"{type(self).__name__}.rename()'s `maxdepth` keyword argument is" " deprecated and will be removed in future universal-pathlib versions.", DeprecationWarning, stacklevel=2, ) kwargs["maxdepth"] = maxdepth self.fs.mv( source_abs.path, target_abs.path, **kwargs, ) return target def replace(self, target: WritablePathLike) -> Self: """ Rename this path to the target path, overwriting if that path exists. The target path may be absolute or relative. Relative paths are interpreted relative to the current working directory, *not* the directory of the Path object. Returns the new Path instance pointing to the target path. Warning ------- This method is currently not implemented. """ _raise_unsupported(type(self).__name__, "replace") @property def drive(self) -> str: """The drive prefix (letter or UNC path), if any. Info ---- On non-Windows systems, the drive is always an empty string. On cloud storage systems, the drive is the bucket name or equivalent. """ if self._relative_base is not None: return "" return self.parser.splitroot(str(self))[0] @property def root(self) -> str: """The root of the path, if any.""" if self._relative_base is not None: return "" return self.parser.splitroot(str(self))[1] def __reduce__(self): 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 def from_uri(cls, uri: str, **storage_options: Any) -> Self: return cls(uri, **storage_options) def as_uri(self) -> str: """Return the string representation of the path as a URI.""" if self._relative_base is not None: raise ValueError( f"relative path can't be expressed as a {self.protocol} URI" ) return str(self) def as_posix(self) -> str: """Return the string representation of the path with POSIX-style separators.""" return str(self) def samefile(self, other_path) -> bool: st = self.stat() if isinstance(other_path, UPath): other_st = other_path.stat() else: other_st = self.with_segments(other_path).stat() return st == other_st @classmethod def cwd(cls) -> Self: """ Return a new UPath object representing the current working directory. Info ---- None of the fsspec filesystems support a global current working directory, so this method only works for the base UPath class, returning the local current working directory. """ if cls is UPath: # default behavior for UPath.cwd() is to return local cwd return get_upath_class("").cwd() # type: ignore[union-attr,return-value] else: _raise_unsupported(cls.__name__, "cwd") @classmethod def home(cls) -> Self: """ Return a new UPath object representing the user's home directory. Info ---- None of the fsspec filesystems support user home directories, so this method only works for the base UPath class, returning the local user's home directory. """ if cls is UPath: return get_upath_class("").home() # type: ignore[union-attr,return-value] else: _raise_unsupported(cls.__name__, "home") def relative_to( # type: ignore[override] self, other: Self | str, /, *_deprecated: Any, walk_up: bool = False, ) -> Self: """Return the relative path to another path identified by the passed arguments. If the operation is not possible (because this is not related to the other path), raise ValueError. The *walk_up* parameter controls whether `..` may be used to resolve the path. """ if walk_up: raise NotImplementedError("walk_up=True is not implemented yet") if isinstance(other, UPath): # revisit: ... if self.__class__ is not other.__class__: raise ValueError( "incompatible protocols:" f" {self.protocol!r} != {other.protocol!r}" ) if self.storage_options != other.storage_options: raise ValueError( "incompatible storage_options:" f" {self.storage_options!r} != {other.storage_options!r}" ) elif isinstance(other, str): other = self.with_segments(other) else: raise TypeError(f"expected UPath or str, got {type(other).__name__}") if other not in self.parents and self != other: raise ValueError(f"{self!s} is not in the subpath of {other!s}") else: rel = copy(self) rel._relative_base = other.path return rel def is_relative_to( self, other: Self | str, /, *_deprecated: Any, ) -> bool: # type: ignore[override] """Return True if the path is relative to another path identified.""" if isinstance(other, UPath) and self.storage_options != other.storage_options: return False elif isinstance(other, str): other = self.with_segments(other) return self == other or other in self.parents def hardlink_to(self, target: ReadablePathLike) -> None: _raise_unsupported(type(self).__name__, "hardlink_to") def full_match( self, pattern: str | SupportsPathLike, *, case_sensitive: bool | None = None, ) -> bool: """Match this path against the provided glob-style pattern. Return True if matching is successful, False otherwise. """ if case_sensitive is not None: warnings.warn( f"{type(self).__name__}.full_match(): case_sensitive" " is currently ignored.", UserWarning, stacklevel=2, ) return super().full_match(str(pattern)) def match( self, path_pattern: str | SupportsPathLike, *, case_sensitive: bool | None = None, ) -> bool: """Match this path against the provided non-recursive glob-style pattern. Return True if matching is successful, False otherwise. """ path_pattern = str(path_pattern) if not path_pattern: raise ValueError("pattern cannot be empty") if case_sensitive is not None: warnings.warn( f"{type(self).__name__}.match(): case_sensitive is currently ignored.", UserWarning, stacklevel=2, ) return self.full_match(path_pattern.replace("**", "*")) @classmethod def __get_pydantic_core_schema__( cls, _source_type: Any, _handler: GetCoreSchemaHandler ) -> CoreSchema: from pydantic_core import core_schema str_schema = core_schema.chain_schema( [ core_schema.str_schema(), core_schema.no_info_plain_validator_function( lambda path: { "path": path, "protocol": None, "storage_options": {}, }, ), ] ) object_schema = core_schema.typed_dict_schema( { "path": core_schema.typed_dict_field( core_schema.str_schema(), required=True ), "protocol": core_schema.typed_dict_field( core_schema.with_default_schema( core_schema.nullable_schema( core_schema.str_schema(), ), default=None, ), required=False, ), "storage_options": core_schema.typed_dict_field( core_schema.with_default_schema( core_schema.dict_schema( core_schema.str_schema(), core_schema.any_schema(), ), default_factory=dict, ), required=False, ), }, extra_behavior="forbid", ) deserialization_schema = core_schema.chain_schema( [ core_schema.union_schema([str_schema, object_schema]), core_schema.no_info_plain_validator_function( lambda dct: cls( dct.pop("path"), protocol=dct.pop("protocol"), **dct["storage_options"], ) ), ] ) serialization_schema = core_schema.plain_serializer_function_ser_schema( lambda u: { "path": u.path, "protocol": u.protocol, "storage_options": dict(u.storage_options), } ) return core_schema.json_or_python_schema( json_schema=deserialization_schema, python_schema=core_schema.union_schema( [core_schema.is_instance_schema(UPath), deserialization_schema] ), serialization=serialization_schema, )