Skip to content

Commit

Permalink
upath: various typing fixes
Browse files Browse the repository at this point in the history
  • Loading branch information
ap-- committed Apr 1, 2024
1 parent 1110dbd commit abede49
Show file tree
Hide file tree
Showing 2 changed files with 95 additions and 46 deletions.
139 changes: 94 additions & 45 deletions upath/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,10 @@
from typing import TYPE_CHECKING
from typing import Any
from typing import BinaryIO
from typing import Generator
from typing import Literal
from typing import Mapping
from typing import Sequence
from typing import TextIO
from typing import TypeVar
from typing import overload
Expand All @@ -22,7 +24,6 @@
else:
from typing_extensions import Self


from fsspec.registry import get_filesystem_class
from fsspec.spec import AbstractFileSystem

Expand All @@ -38,6 +39,9 @@
from upath._stat import UPathStatResult
from upath.registry import get_upath_class

if TYPE_CHECKING:
from urllib.parse import SplitResult

__all__ = ["UPath"]


Expand Down Expand Up @@ -101,7 +105,24 @@ class UPath(PathlibPathShim, Path):
"__root",
"__parts",
)

if TYPE_CHECKING:
# public
anchor: str
drive: str
parent: Self
parents: Sequence[Self]
parts: tuple[str, ...]
root: str
stem: str
suffix: str
suffixes: list[str]

def with_name(self, name: str) -> Self: ...
def with_stem(self, stem: str) -> Self: ...
def with_suffix(self, suffix: str) -> Self: ...

# private attributes
_protocol: str
_storage_options: dict[str, Any]
_fs_cached: AbstractFileSystem
Expand Down Expand Up @@ -416,7 +437,7 @@ def _kwargs(self):
return self.storage_options

@property
def _url(self):
def _url(self) -> SplitResult:
# TODO:
# _url should be deprecated, but for now there is no good way of
# accessing query parameters from urlpaths...
Expand Down Expand Up @@ -688,15 +709,21 @@ def __hash__(self) -> int:
"""
return hash((self.protocol, self.path))

def relative_to(self, other, /, *_deprecated, walk_up=False):
def relative_to( # type: ignore[override]
self,
other,
/,
*_deprecated,
walk_up=False,
) -> Self:
if isinstance(other, UPath) and self.storage_options != other.storage_options:
raise ValueError(
"paths have different storage_options:"
f" {self.storage_options!r} != {other.storage_options!r}"
)
return super().relative_to(other, *_deprecated, walk_up=walk_up)

def is_relative_to(self, other, /, *_deprecated):
def is_relative_to(self, other, /, *_deprecated) -> bool: # type: ignore[override]
if isinstance(other, UPath) and self.storage_options != other.storage_options:
return False
return super().is_relative_to(other, *_deprecated)
Expand Down Expand Up @@ -728,23 +755,23 @@ def stat( # type: ignore[override]
)
return UPathStatResult.from_info(self.fs.stat(self.path))

def lstat(self):
def lstat(self) -> UPathStatResult: # type: ignore[override]
# return self.stat(follow_symlinks=False)
raise NotImplementedError

def exists(self, *, follow_symlinks=True):
def exists(self, *, follow_symlinks=True) -> bool:
return self.fs.exists(self.path)

def is_dir(self):
def is_dir(self) -> bool:
return self.fs.isdir(self.path)

def is_file(self):
def is_file(self) -> bool:
return self.fs.isfile(self.path)

def is_mount(self):
def is_mount(self) -> bool:
return False

def is_symlink(self):
def is_symlink(self) -> bool:
try:
info = self.fs.info(self.path)
if "islink" in info:
Expand All @@ -753,22 +780,22 @@ def is_symlink(self):
return False
return False

def is_junction(self):
def is_junction(self) -> bool:
return False

def is_block_device(self):
def is_block_device(self) -> bool:
return False

def is_char_device(self):
def is_char_device(self) -> bool:
return False

def is_fifo(self):
def is_fifo(self) -> bool:
return False

def is_socket(self):
def is_socket(self) -> bool:
return False

def samefile(self, other_path):
def samefile(self, other_path) -> bool:
raise NotImplementedError

@overload # type: ignore[override]
Expand All @@ -783,15 +810,15 @@ def open(
) -> TextIO: ...

@overload
def open(
def open( # type: ignore[override]
self,
mode: Literal["rb", "wb", "ab"],
buffering: int = ...,
encoding: str = ...,
errors: str = ...,
newline: str = ...,
**fsspec_kwargs: Any,
) -> BinaryIO: ... # type: ignore[override]
) -> BinaryIO: ...

def open(
self,
Expand Down Expand Up @@ -830,7 +857,7 @@ def open(
fsspec_kwargs.setdefault("block_size", fsspec_kwargs.pop("buffering"))
return self.fs.open(self.path, mode=mode, **fsspec_kwargs)

def iterdir(self):
def iterdir(self) -> Generator[UPath, None, None]:
for name in self.fs.listdir(self.path):
# fsspec returns dictionaries
if isinstance(name, dict):
Expand All @@ -850,15 +877,19 @@ def _make_child_relpath(self, name):
del path._str # fix _str = str(self) assignment
return path

def glob(self, pattern: str, *, case_sensitive=None):
def glob(
self, pattern: str, *, case_sensitive=None
) -> Generator[UPath, None, None]:
path_pattern = self.joinpath(pattern).path
sep = self._flavour.sep
base = self.fs._strip_protocol(self.path)
for name in self.fs.glob(path_pattern):
name = str_remove_prefix(str_remove_prefix(name, base), sep)
yield self.joinpath(name)

def rglob(self, pattern: str, *, case_sensitive=None):
def rglob(
self, pattern: str, *, case_sensitive=None
) -> Generator[UPath, None, None]:
if _FSSPEC_HAS_WORKING_GLOB is None:
_check_fsspec_has_working_glob()

Expand Down Expand Up @@ -886,23 +917,23 @@ def rglob(self, pattern: str, *, case_sensitive=None):
yield self.joinpath(name)

@classmethod
def cwd(cls):
def cwd(cls) -> UPath:
if cls is UPath:
return get_upath_class("").cwd()
return get_upath_class("").cwd() # type: ignore[union-attr]
else:
raise NotImplementedError

@classmethod
def home(cls):
def home(cls) -> UPath:
if cls is UPath:
return get_upath_class("").home()
return get_upath_class("").home() # type: ignore[union-attr]
else:
raise NotImplementedError

def absolute(self):
def absolute(self) -> Self:
return self

def resolve(self, strict: bool = False):
def resolve(self, strict: bool = False) -> Self:
_parts = self.parts

# Do not attempt to normalize path if no parts are dots
Expand All @@ -920,19 +951,19 @@ def resolve(self, strict: bool = False):

return self.with_segments(*_parts[:1], *resolved)

def owner(self):
def owner(self) -> str:
raise NotImplementedError

def group(self):
def group(self) -> str:
raise NotImplementedError

def readlink(self):
def readlink(self) -> Self:
raise NotImplementedError

def touch(self, mode=0o666, exist_ok=True):
def touch(self, mode=0o666, exist_ok=True) -> None:
self.fs.touch(self.path, truncate=not exist_ok)

def mkdir(self, mode=0o777, parents=False, exist_ok=False):
def mkdir(self, mode=0o777, parents=False, exist_ok=False) -> None:
if parents and not exist_ok and self.exists():
raise FileExistsError(str(self))
try:
Expand All @@ -947,45 +978,63 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False):
if not self.is_dir():
raise FileExistsError(str(self))

def chmod(self, mode, *, follow_symlinks=True):
def chmod(self, mode: int, *, follow_symlinks: bool = True) -> None:
raise NotImplementedError

def lchmod(self, mode: int) -> None:
raise NotImplementedError

def unlink(self, missing_ok=False):
def unlink(self, missing_ok: bool = False) -> None:
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): # fixme: non-standard
def rmdir(self, recursive: bool = True) -> None: # fixme: non-standard
if not self.is_dir():
raise NotADirectoryError(str(self))
if not recursive and next(self.iterdir()):
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, *, recursive=False, maxdepth=None, **kwargs
): # fixme: non-standard
self,
target: str | os.PathLike[str] | UPath,
*,
recursive: bool = False,
maxdepth: int | None = None,
**kwargs: Any,
) -> UPath: # fixme: non-standard
target_: UPath
if not isinstance(target, UPath):
target = self.parent.joinpath(target).resolve()
target_ = self.parent.joinpath(target).resolve()
else:
target_ = target
self.fs.mv(
self.path,
target.path,
target_.path,
recursive=recursive,
maxdepth=maxdepth,
**kwargs,
)
return target
return target_

def replace(self, target):
def replace(self, target: str | os.PathLike[str] | UPath) -> UPath:
raise NotImplementedError # todo

def symlink_to(self, target, target_is_directory=False):
def symlink_to( # type: ignore[override]
self,
target: str | os.PathLike[str] | UPath,
target_is_directory: bool = False,
) -> None:
raise NotImplementedError

def hardlink_to(self, target):
def hardlink_to( # type: ignore[override]
self,
target: str | os.PathLike[str] | UPath,
) -> None:
raise NotImplementedError

def expanduser(self):
def expanduser(self) -> Self:
raise NotImplementedError
2 changes: 1 addition & 1 deletion upath/implementations/http.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ def _transform_init_args(
return args, protocol, storage_options

@property
def root(self) -> str:
def root(self) -> str: # type: ignore[override]
return super().root or "/"

def __str__(self):
Expand Down

0 comments on commit abede49

Please sign in to comment.