Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 14 additions & 16 deletions upath/_flavour.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
from typing import TYPE_CHECKING
from typing import Any
from typing import TypedDict
from typing import Union
from urllib.parse import SplitResult
from urllib.parse import urlsplit

Expand All @@ -22,7 +21,7 @@
from upath._flavour_sources import flavour_registry
from upath._protocol import get_upath_protocol
from upath._protocol import normalize_empty_netloc
from upath.types import JoinablePath
from upath.types import JoinablePathLike
from upath.types import UPathParser

if TYPE_CHECKING:
Expand All @@ -42,7 +41,6 @@
]

class_registry: Mapping[str, type[AbstractFileSystem]] = _class_registry
PathOrStr: TypeAlias = Union[str, os.PathLike[str], JoinablePath]


class AnyProtocolFileSystemFlavour(FileSystemFlavourBase):
Expand Down Expand Up @@ -237,7 +235,7 @@ def local_file(self) -> bool:
return bool(getattr(self._spec, "local_file", False))

@staticmethod
def stringify_path(pth: PathOrStr) -> str:
def stringify_path(pth: JoinablePathLike) -> str:
if isinstance(pth, str):
out = pth
elif isinstance(pth, upath.UPath) and not pth.is_absolute():
Expand All @@ -253,18 +251,18 @@ def stringify_path(pth: PathOrStr) -> str:
out = str(pth)
return normalize_empty_netloc(out)

def strip_protocol(self, pth: PathOrStr) -> str:
def strip_protocol(self, pth: JoinablePathLike) -> str:
pth = self.stringify_path(pth)
return self._spec._strip_protocol(pth)

def get_kwargs_from_url(self, url: PathOrStr) -> dict[str, Any]:
def get_kwargs_from_url(self, url: JoinablePathLike) -> dict[str, Any]:
# NOTE: the public variant is _from_url not _from_urls
if hasattr(url, "storage_options"):
return dict(url.storage_options)
url = self.stringify_path(url)
return self._spec._get_kwargs_from_urls(url)

def parent(self, path: PathOrStr) -> str:
def parent(self, path: JoinablePathLike) -> str:
path = self.stringify_path(path)
return self._spec._parent(path)

Expand All @@ -278,14 +276,14 @@ def sep(self) -> str: # type: ignore[override]
def altsep(self) -> str | None: # type: ignore[override]
return None

def isabs(self, path: PathOrStr) -> bool:
def isabs(self, path: JoinablePathLike) -> bool:
path = self.strip_protocol(path)
if self.local_file:
return os.path.isabs(path)
else:
return path.startswith(self.root_marker)

def join(self, path: PathOrStr, *paths: PathOrStr) -> str:
def join(self, path: JoinablePathLike, *paths: JoinablePathLike) -> str:
if not paths:
return self.strip_protocol(path) or self.root_marker
if self.local_file:
Expand All @@ -309,7 +307,7 @@ def join(self, path: PathOrStr, *paths: PathOrStr) -> str:
else:
return drv + posixpath.join(p0, *pN)

def split(self, path: PathOrStr):
def split(self, path: JoinablePathLike) -> tuple[str, str]:
stripped_path = self.strip_protocol(path)
if self.local_file:
return os.path.split(stripped_path)
Expand All @@ -328,7 +326,7 @@ def split(self, path: PathOrStr):
return self.split(head)
return head, tail

def splitdrive(self, path: PathOrStr) -> tuple[str, str]:
def splitdrive(self, path: JoinablePathLike) -> tuple[str, str]:
path = self.strip_protocol(path)
if self.netloc_is_anchor:
u = urlsplit(path)
Expand All @@ -353,13 +351,13 @@ def splitdrive(self, path: PathOrStr) -> tuple[str, str]:
# all other cases don't have a drive
return "", path

def normcase(self, path: PathOrStr) -> str:
def normcase(self, path: JoinablePathLike) -> str:
if self.local_file:
return os.path.normcase(self.stringify_path(path))
else:
return self.stringify_path(path)

def splitext(self, path: PathOrStr) -> tuple[str, str]:
def splitext(self, path: JoinablePathLike) -> tuple[str, str]:
path = self.stringify_path(path)
if self.local_file:
return os.path.splitext(path)
Expand All @@ -375,7 +373,7 @@ def splitext(self, path: PathOrStr) -> tuple[str, str]:

# === Python3.12 pathlib flavour ==================================

def splitroot(self, path: PathOrStr) -> tuple[str, str, str]:
def splitroot(self, path: JoinablePathLike) -> tuple[str, str, str]:
drive, tail = self.splitdrive(path)
if self.netloc_is_anchor:
root_marker = self.root_marker or self.sep
Expand Down Expand Up @@ -422,13 +420,13 @@ def __repr__(self):
return f"<{cls_name} of {self._owner.__name__}>"


def upath_strip_protocol(pth: PathOrStr) -> str:
def upath_strip_protocol(pth: JoinablePathLike) -> str:
if protocol := get_upath_protocol(pth):
return WrappedFileSystemFlavour.from_protocol(protocol).strip_protocol(pth)
return WrappedFileSystemFlavour.stringify_path(pth)


def upath_get_kwargs_from_url(url: PathOrStr) -> dict[str, Any]:
def upath_get_kwargs_from_url(url: JoinablePathLike) -> dict[str, Any]:
if protocol := get_upath_protocol(url):
return WrappedFileSystemFlavour.from_protocol(protocol).get_kwargs_from_url(url)
return {}
Expand Down
9 changes: 5 additions & 4 deletions upath/_protocol.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from __future__ import annotations

import os
import re
from collections import ChainMap
from pathlib import PurePath
Expand All @@ -11,7 +10,7 @@
from fsspec.registry import registry as _registry

if TYPE_CHECKING:
from upath.types import JoinablePath
from upath.types import JoinablePathLike

__all__ = [
"get_upath_protocol",
Expand Down Expand Up @@ -60,7 +59,7 @@ def _fsspec_protocol_equals(p0: str, p1: str) -> bool:


def get_upath_protocol(
pth: str | os.PathLike[str] | PurePath | JoinablePath,
pth: JoinablePathLike,
*,
protocol: str | None = None,
storage_options: dict[str, Any] | None = None,
Expand All @@ -74,6 +73,8 @@ def get_upath_protocol(
pth_protocol = pth.protocol
elif isinstance(pth, PurePath):
pth_protocol = getattr(pth, "protocol", "")
elif hasattr(pth, "__vfspath__"):
pth_protocol = _match_protocol(pth.__vfspath__())
elif hasattr(pth, "__fspath__"):
pth_protocol = _match_protocol(pth.__fspath__())
else:
Expand Down Expand Up @@ -102,7 +103,7 @@ def normalize_empty_netloc(pth: str) -> str:

def compatible_protocol(
protocol: str,
*args: str | os.PathLike[str] | PurePath | JoinablePath,
*args: JoinablePathLike,
) -> bool:
"""check if UPath protocols are compatible"""
from upath.core import UPath
Expand Down
5 changes: 3 additions & 2 deletions upath/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,10 +38,11 @@
from upath.registry import get_upath_class
from upath.types import UNSET_DEFAULT
from upath.types import JoinablePathLike
from upath.types import OpenablePath
from upath.types import PathInfo
from upath.types import ReadablePath
from upath.types import ReadablePathLike
from upath.types import UPathParser
from upath.types import WritablePath
from upath.types import WritablePathLike

if TYPE_CHECKING:
Expand Down Expand Up @@ -427,7 +428,7 @@ def _url(self) -> SplitResult:
return urlsplit(self.__str__())


class UPath(_UPathMixin, OpenablePath):
class UPath(_UPathMixin, WritablePath, ReadablePath):
__slots__ = (
"_chain",
"_chain_parser",
Expand Down
16 changes: 0 additions & 16 deletions upath/tests/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,6 @@
from upath import UPath
from upath.implementations.cloud import GCSPath
from upath.implementations.cloud import S3Path
from upath.types import CompatOpenablePath
from upath.types import CompatReadablePath
from upath.types import CompatWritablePath
from upath.types import OpenablePath
from upath.types import ReadablePath
from upath.types import WritablePath

Expand Down Expand Up @@ -123,10 +119,6 @@ def test_subclass_with_gcs():
assert isinstance(path, UPath)
assert isinstance(path, ReadablePath)
assert isinstance(path, WritablePath)
assert isinstance(path, OpenablePath)
assert isinstance(path, CompatReadablePath)
assert isinstance(path, CompatWritablePath)
assert isinstance(path, CompatOpenablePath)
assert not isinstance(path, os.PathLike)
assert not isinstance(path, pathlib.Path)

Expand All @@ -137,10 +129,6 @@ def test_instance_check(local_testdir):
assert isinstance(path, UPath)
assert isinstance(path, ReadablePath)
assert isinstance(path, WritablePath)
assert isinstance(path, OpenablePath)
assert isinstance(path, CompatReadablePath)
assert isinstance(path, CompatWritablePath)
assert isinstance(path, CompatOpenablePath)
assert isinstance(path, os.PathLike)
assert isinstance(path, pathlib.Path)

Expand All @@ -150,10 +138,6 @@ def test_instance_check_local_uri(local_testdir):
assert isinstance(path, UPath)
assert isinstance(path, ReadablePath)
assert isinstance(path, WritablePath)
assert isinstance(path, OpenablePath)
assert isinstance(path, CompatReadablePath)
assert isinstance(path, CompatWritablePath)
assert isinstance(path, CompatOpenablePath)
assert isinstance(path, os.PathLike)
assert not isinstance(path, pathlib.Path)

Expand Down
Loading