upath: more typing fixes

upath/_compat.py

@@ -490,7 +490,8 @@ def mv(self, path, target, recursive=False, maxdepth=None, **kwargs):
         )
 
 
-F = TypeVar("F")
+RT = TypeVar("RT")
+F = Callable[..., RT]
 
 
 def deprecated(*, python_version: tuple[int, ...]) -> Callable[[F], F]:

upath/_flavour.py

@@ -19,7 +19,7 @@
     TypeAlias = Any
 
 from fsspec.registry import known_implementations
-from fsspec.registry import registry as class_registry
+from fsspec.registry import registry as _class_registry
 from fsspec.spec import AbstractFileSystem
 
 from upath._compat import deprecated
@@ -40,14 +40,14 @@
     "upath_get_kwargs_from_url",
 ]
 
-class_registry: Mapping[str, type[AbstractFileSystem]]
+class_registry: Mapping[str, type[AbstractFileSystem]] = _class_registry
 PathOrStr: TypeAlias = Union[str, "os.PathLike[str]"]
 
 
 class AnyProtocolFileSystemFlavour(FileSystemFlavourBase):
-    sep: str = "/"
-    protocol: tuple[str, ...] = ()
-    root_marker: str = "/"
+    sep = "/"
+    protocol = ()
+    root_marker = "/"
 
     @classmethod
     def _strip_protocol(cls, path: str) -> str:
@@ -167,10 +167,11 @@ def from_protocol(
     ) -> WrappedFileSystemFlavour:
         """return the fsspec flavour for the given protocol"""
 
+        _c = cls.protocol_config
         config = {
-            key: True
-            for key, protocols in cls.protocol_config.items()
-            if protocol in protocols
+            "netloc_is_anchor": protocol in _c["netloc_is_anchor"],
+            "supports_empty_parts": protocol in _c["supports_empty_parts"],
+            "meaningful_trailing_slash": protocol in _c["meaningful_trailing_slash"],
         }
 
         # first try to get an already imported fsspec filesystem class
@@ -232,10 +233,6 @@ def stringify_path(pth: PathOrStr) -> str:
             out = str(pth)
         return normalize_empty_netloc(out)
 
-    def empty_part_join(self, path: str, *paths: str) -> str:
-        sep = self.sep
-        return sep.join([str_remove_suffix(path, sep), *paths])
-
     def strip_protocol(self, pth: PathOrStr) -> str:
         pth = self.stringify_path(pth)
         return self._spec._strip_protocol(pth)
@@ -269,21 +266,23 @@ def isabs(self, path: PathOrStr) -> bool:
             return path.startswith(self.root_marker)
 
     def join(self, path: PathOrStr, *paths: PathOrStr) -> str:
-        if self.supports_empty_parts:
-            _join = self.empty_part_join
-        else:
-            _join = posixpath.join
         if self.netloc_is_anchor:
             drv, p0 = self.splitdrive(path)
             pN = list(map(self.stringify_path, paths))
             if not drv and not p0:
                 path, *pN = pN
                 drv, p0 = self.splitdrive(path)
-            return drv + _join(p0 or self.sep, *pN)
+            p0 = p0 or self.sep
         else:
             p0 = str(self.strip_protocol(path))
-            pN = map(self.stringify_path, paths)
-            return _join(p0, *pN)
+            pN = list(map(self.stringify_path, paths))
+            drv = ""
+        if self.supports_empty_parts:
+            return drv + self.sep.join(
+                [str_remove_suffix(p0, self.sep), *pN]
+            )
+        else:
+            return drv + posixpath.join(p0, *pN)
 
     def split(self, path: PathOrStr):
         stripped_path = self.strip_protocol(path)
@@ -384,20 +383,21 @@ class LazyFlavourDescriptor:
     """descriptor to lazily get the flavour for a given protocol"""
 
     def __init__(self) -> None:
-        self._owner = None
+        self._owner: type[UPath] | None = None
 
     def __set_name__(self, owner: type[UPath], name: str) -> None:
         # helper to provide a more informative repr
         self._owner = owner
+        self._default_protocol: str | None
         try:
-            self._default_protocol = self._owner.protocols[0]
+            self._default_protocol = self._owner.protocols[0]  # type: ignore
         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:
+        elif self._default_protocol:  # type: ignore
             return WrappedFileSystemFlavour.from_protocol(self._default_protocol)
         else:
             return default_flavour
@@ -464,7 +464,7 @@ def upath_urijoin(base: str, uri: str) -> str:
         segments = base_parts + us.path.split("/")
         segments[1:-1] = filter(None, segments[1:-1])
 
-    resolved_path = []
+    resolved_path: list[str] = []
 
     for seg in segments:
         if seg == "..":

upath/_stat.py

@@ -45,7 +45,7 @@ def _get_stat_result_extra_fields() -> tuple[str, ...]:
     sr = os.stat_result(range(os.stat_result.n_fields))
     rd = sr.__reduce__()
     assert isinstance(rd, tuple), "unexpected return os.stat_result.__reduce__"
-    _, (_, extra) = sr.__reduce__()
+    _, (_, extra) = rd
     extra_fields = sorted(extra, key=extra.__getitem__)
     return tuple(extra_fields)
 
@@ -317,7 +317,7 @@ def __iter__(self) -> Iterator[int]:
         for field in self._fields:
             yield int(getattr(self, field))
 
-    def index(self, value: int, start: int = 0, stop: int = None, /) -> int:
+    def index(self, value: int, start: int = 0, stop: int | None = None, /) -> int:
         """the sequence interface index method."""
         if stop is None:
             stop = len(self._seq)

upath/core.py

@@ -693,7 +693,7 @@ def name(self):
 
     # === pathlib.Path ================================================
 
-    def stat(self, *, follow_symlinks=True) -> UPathStatResult:
+    def stat(self, *, follow_symlinks=True) -> UPathStatResult:  # type: ignore[override]
         if not follow_symlinks:
             warnings.warn(
                 "UPath.stat(follow_symlinks=False): follow_symlinks=False is"
@@ -746,10 +746,10 @@ def is_socket(self):
     def samefile(self, other_path):
         raise NotImplementedError
 
-    @overload
+    @overload  # type: ignore[override]
     def open(
         self,
-        mode: Literal["r", "w", "a"] = ...,
+        mode: Literal["r", "w", "a"] = "r",
         buffering: int = ...,
         encoding: str = ...,
         errors: str = ...,
@@ -760,13 +760,13 @@ def open(
     @overload
     def open(
         self,
-        mode: Literal["rb", "wb", "ab"] = ...,
+        mode: Literal["rb", "wb", "ab"],
         buffering: int = ...,
         encoding: str = ...,
         errors: str = ...,
         newline: str = ...,
         **fsspec_kwargs: Any,
-    ) -> BinaryIO: ...
+    ) -> BinaryIO: ...  # type: ignore[override]
 
     def open(
         self,

upath/implementations/http.py

@@ -28,7 +28,7 @@ def _transform_init_args(
     ) -> tuple[tuple[str | os.PathLike, ...], str, dict[str, Any]]:
         # allow initialization via a path argument and protocol keyword
         if args and not str(args[0]).startswith(protocol):
-            args = (f"{protocol}://{args[0].lstrip('/')}", *args[1:])
+            args = (f"{protocol}://{str(args[0]).lstrip('/')}", *args[1:])
         return args, protocol, storage_options
 
     @property

upath/implementations/local.py

@@ -105,13 +105,13 @@ def _upath_init(inst: PosixUPath | WindowsUPath) -> None:
         inst._init()
 
 
-class PosixUPath(PosixPath, LocalPath):
+class PosixUPath(PosixPath, LocalPath):  # type: ignore[misc]
     __slots__ = ()
 
     # assign all PosixPath methods/attrs to prevent multi inheritance issues
     _set_class_attributes(locals(), src=PosixPath)
 
-    def open(
+    def open(  # type: ignore[override]
         self,
         mode="r",
         buffering=-1,
@@ -136,14 +136,14 @@ def open(
 
         def __new__(
             cls, *args, protocol: str | None = None, **storage_options: Any
-        ) -> UPath:
+        ) -> PosixUPath:
             if os.name == "nt":
                 raise NotImplementedError(
                     f"cannot instantiate {cls.__name__} on your system"
                 )
             obj = super().__new__(cls, *args)
             obj._protocol = ""
-            return obj
+            return obj  # type: ignore[return-value]
 
         def __init__(
             self, *args, protocol: str | None = None, **storage_options: Any
@@ -169,13 +169,13 @@ def path(self) -> str:
             return PosixPath.__str__(self)
 
 
-class WindowsUPath(WindowsPath, LocalPath):
+class WindowsUPath(WindowsPath, LocalPath):  # type: ignore[misc]
     __slots__ = ()
 
     # assign all WindowsPath methods/attrs to prevent multi inheritance issues
     _set_class_attributes(locals(), src=WindowsPath)
 
-    def open(
+    def open(  # type: ignore[override]
         self,
         mode="r",
         buffering=-1,
@@ -200,14 +200,14 @@ def open(
 
         def __new__(
             cls, *args, protocol: str | None = None, **storage_options: Any
-        ) -> UPath:
+        ) -> WindowsUPath:
             if os.name != "nt":
                 raise NotImplementedError(
                     f"cannot instantiate {cls.__name__} on your system"
                 )
             obj = super().__new__(cls, *args)
             obj._protocol = ""
-            return obj
+            return obj  # type: ignore[return-value]
 
         def __init__(
             self, *args, protocol: str | None = None, **storage_options: Any

upath/tests/conftest.py

@@ -24,7 +24,7 @@ class DummyTestFS(LocalFileSystem):
     root_marker = "/"
 
     @classmethod
-    def _strip_protocol(cls, path):
+    def _strip_protocol(cls, path, **_):
         path = stringify_path(path)
         if path.startswith("mock://"):
             path = path[7:]