lint: fix type hints, pylint issues

Author: pmrowla

pyproject.toml

@@ -70,6 +70,7 @@ files = ["src", "tests"]
 [[tool.mypy.overrides]]
 module = [
     "pygtrie",
+    "dvc_http.*",
     "funcy",
     "git",
     "gitdb.*",

src/scmrepo/git/backend/dulwich/__init__.py

@@ -51,7 +51,7 @@ def __init__(self, repo, name, mode, sha):
         self._mode = mode
         self._sha = sha
 
-    def open(
+    def open(  # pylint: disable=unused-argument
         self,
         mode: str = "r",
         encoding: Optional[str] = None,
@@ -153,18 +153,18 @@ def encoding(self) -> str:
             return self._config.encoding
         return self._config.backends[0].encoding
 
-    def get(self, section: Tuple[str], name: str) -> str:
+    def get(self, section: Tuple[str, ...], name: str) -> str:
         """Return the specified setting as a string."""
         return self._config.get(section, name).decode(self.encoding)
 
-    def get_bool(self, section: Tuple[str], name: str) -> bool:
+    def get_bool(self, section: Tuple[str, ...], name: str) -> bool:
         """Return the specified setting as a boolean."""
         value = self._config.get_boolean(section, name)
         if value is None:
             raise ValueError("setting is not a valid boolean")
         return value
 
-    def get_multivar(self, section: Tuple[str], name: str) -> Iterator[str]:
+    def get_multivar(self, section: Tuple[str, ...], name: str) -> Iterator[str]:
         """Iterate over string values in the specified multivar setting."""
         for value in self._config.get_multivar(section, name):
             yield value.decode(self.encoding)

src/scmrepo/git/backend/pygit2/__init__.py

@@ -6,7 +6,6 @@
 from io import BytesIO, StringIO, TextIOWrapper
 from typing import (
     TYPE_CHECKING,
-    Any,
     Callable,
     Dict,
     Generator,
@@ -55,7 +54,7 @@ def open(
         self,
         mode: str = "r",
         encoding: str = None,
-        key: tuple = None,
+        key: Optional[Tuple[str, ...]] = None,
         raw: bool = True,
         rev: Optional[str] = None,
         **kwargs,
@@ -67,14 +66,16 @@ def open(
         if self.backend is not None:
             try:
                 if rev:
+                    # pylint: disable-next=protected-access
                     commit, _ref = self.backend._resolve_refish(rev)
                 else:
                     pass
                 if raw:
                     blob_kwargs = {}
                 else:
+                    assert key is not None
                     path = "/".join(key)
-                    blob_kwargs: Dict[str, Any] = {
+                    blob_kwargs = {
                         "as_path": path,
                         "commit_id": commit.oid,
                     }
@@ -123,21 +124,21 @@ class Pygit2Config(Config):
     def __init__(self, config: "_Pygit2Config"):
         self._config = config
 
-    def _key(self, section: Tuple[str], name: str) -> str:
+    def _key(self, section: Tuple[str, ...], name: str) -> str:
         return ".".join(section + (name,))
 
-    def get(self, section: Tuple[str], name: str) -> str:
+    def get(self, section: Tuple[str, ...], name: str) -> str:
         return self._config[self._key(section, name)]
 
-    def get_bool(self, section: Tuple[str], name: str) -> bool:
+    def get_bool(self, section: Tuple[str, ...], name: str) -> bool:
         from pygit2 import GitError
 
         try:
             return self._config.get_bool(self._key(section, name))
         except GitError as exc:
             raise ValueError("invalid boolean config entry") from exc
 
-    def get_multivar(self, section: Tuple[str], name: str) -> Iterator[str]:
+    def get_multivar(self, section: Tuple[str, ...], name: str) -> Iterator[str]:
         from pygit2 import GitError
 
         try:

src/scmrepo/git/backend/pygit2/filter.py

@@ -15,7 +15,7 @@ class LFSFilter(Filter):
 
     def __init__(self, *args, **kwargs):
         self._smudge_buf: Optional[io.BytesIO] = None
-        self._smudge_git_dir: Optional[str] = None
+        self._smudge_root: Optional[str] = None
 
     def check(self, src: "FilterSource", attr_values: List[str]):
         if attr_values[0] == "lfs":
@@ -48,6 +48,7 @@ def _smudge(self, write_next: Callable[[bytes], None]):
         from scmrepo.git.lfs import smudge
         from scmrepo.git.lfs.fetch import get_fetch_url
 
+        assert self._smudge_buf is not None
         self._smudge_buf.seek(0)
         with Git(self._smudge_root) as scm:
             try:

src/scmrepo/git/config.py

@@ -10,15 +10,15 @@ class Config(ABC):
     """Read-only Git config."""
 
     @abstractmethod
-    def get(self, section: Tuple[str], name: str) -> str:
+    def get(self, section: Tuple[str, ...], name: str) -> str:
         """Return the specified setting as a string.
 
         Raises:
             KeyError: Option was not set.
         """
 
     @abstractmethod
-    def get_bool(self, section: Tuple[str], name: str) -> bool:
+    def get_bool(self, section: Tuple[str, ...], name: str) -> bool:
         """Return the specified setting as a boolean.
 
         Raises:
@@ -27,7 +27,7 @@ def get_bool(self, section: Tuple[str], name: str) -> bool:
         """
 
     @abstractmethod
-    def get_multivar(self, section: Tuple[str], name: str) -> Iterator[str]:
+    def get_multivar(self, section: Tuple[str, ...], name: str) -> Iterator[str]:
         """Iterate over string values in the specified multivar setting.
 
         Raises:

src/scmrepo/git/lfs/client.py

@@ -1,7 +1,7 @@
 import logging
 from contextlib import AbstractContextManager
 from functools import wraps
-from typing import TYPE_CHECKING, Any, Coroutine, Dict, Iterable, Optional
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Iterable, Optional
 
 import aiohttp
 from dvc_http import HTTPFileSystem
@@ -27,9 +27,10 @@
 
 class _LFSClient(ReadOnlyRetryClient):
     async def _request(self, *args, **kwargs):
-        return await super()._request(*args, **kwargs)
+        return await super()._request(*args, **kwargs)  # pylint: disable=no-member
 
 
+# pylint: disable=abstract-method
 class _LFSFileSystem(HTTPFileSystem):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -71,10 +72,11 @@ async def get_client(self, **kwargs):
         return ReadOnlyRetryClient(**kwargs)
 
 
-def _authed(f: Coroutine):
+def _authed(f: Callable[..., Awaitable]):
     """Set credentials and retry the given coroutine if needed."""
 
-    @wraps(f)
+    # pylint: disable=protected-access
+    @wraps(f)  # type: ignore[arg-type]
     async def wrapper(self, *args, **kwargs):
         try:
             return await f(self, *args, **kwargs)
@@ -110,7 +112,7 @@ def __init__(
         """
         self.url = url
         self.git_url = git_url
-        self.headers = {}
+        self.headers: Dict[str, str] = headers or {}
 
     def __exit__(self, *args, **kwargs):
         self.close()

src/scmrepo/git/lfs/fetch.py

@@ -1,7 +1,7 @@
 import fnmatch
 import io
 import os
-from typing import TYPE_CHECKING, Callable, Iterable, Iterator, List, Optional
+from typing import TYPE_CHECKING, Callable, Iterable, Iterator, List, Optional, Set
 
 from scmrepo.exceptions import InvalidRemote, SCMError
 
@@ -24,7 +24,7 @@ def fetch(
     # NOTE: This currently does not support fetching objects from the worktree
     if not revs:
         revs = ["HEAD"]
-    objects = set()
+    objects: Set[Pointer] = set()
     for rev in revs:
         objects.update(
             pointer
@@ -82,6 +82,7 @@ def get_fetch_url(scm: "Git", remote: Optional[str] = None):  # noqa: C901
             remote = "origin"
 
     # check remote.*.lfsurl (can be set in git config and .lfsconfig)
+    assert remote is not None
     try:
         return git_config.get(("remote", remote), "lfsurl")
     except KeyError:
@@ -156,6 +157,6 @@ def _filter_paths(
         help="Refs or commits to fetch. Defaults to 'HEAD'.",
     )
     args = parser.parse_args()
-    with Git(".") as scm:
+    with Git(".") as scm_:  # pylint: disable=E0601
         print("fetch: fetching reference", ", ".join(args.refs), file=sys.stderr)
-        fetch(scm, revs=args.refs, remote=args.remote)
+        fetch(scm_, revs=args.refs, remote=args.remote)

src/scmrepo/git/lfs/pointer.py

@@ -14,7 +14,8 @@
 
 
 def _get_kv(line: str) -> Tuple[str, str]:
-    return line.strip().split(maxsplit=1)
+    key, value = line.strip().split(maxsplit=1)
+    return key, value
 
 
 @dataclass
@@ -41,8 +42,9 @@ def build(cls, fobj: BinaryIO) -> "Pointer":
     def load(cls, fobj: IO) -> "Pointer":
         """Load the specified pointer file."""
 
-        if isinstance(fobj, io.TextIOBase):
-            text_obj: TextIO = fobj
+        if isinstance(fobj, io.TextIOBase):  # type: ignore[unreachable]
+            text_obj: TextIO = fobj  # type: ignore[unreachable]
+
         else:
             text_obj = io.TextIOWrapper(fobj, encoding="utf-8")
 
@@ -102,6 +104,6 @@ def to_bytes(self) -> bytes:
         sys.exit("Nothing to do")
 
     print(f"Git LFS pointer for {args.file}\n", file=sys.stderr)
-    with open(args.file, mode="rb") as fobj:
-        p = Pointer.build(fobj)
+    with open(args.file, mode="rb") as fobj_:
+        p = Pointer.build(fobj_)
     print(p.dump(), end="")

src/scmrepo/git/lfs/progress.py

@@ -35,13 +35,13 @@ def _update_git(self):
 
     def branch(
         self,
-        path_1: "Union[str, BinaryIO]",
+        path_1: Union[str, BinaryIO],
         path_2: str,
         kwargs: Dict[str, Any],
         child: Optional[Callback] = None,
     ):
         if child:
-            child = child
+            pass
         elif self.git_progress:
             child = TqdmCallback(
                 bytes=True, desc=path_1 if isinstance(path_1, str) else path_2

src/scmrepo/git/lfs/smudge.py

@@ -14,7 +14,7 @@ def smudge(
     storage: "LFSStorage", fobj: BinaryIO, url: Optional[str] = None
 ) -> BinaryIO:
     """Wrap the specified binary IO stream and run LFS smudge if necessary."""
-    reader = io.BufferedReader(fobj)
+    reader = io.BufferedReader(fobj)  # type: ignore[arg-type]
     data = reader.peek(100)
     if any(data.startswith(header) for header in HEADERS):
         # read the pointer data into memory since the raw stream is unseekable
@@ -45,7 +45,7 @@ def smudge(
         )
     scm = Git()
     try:
-        with smudge(scm.lfs_storage, sys.stdin.buffer) as fobj:
-            sys.stdout.buffer.write(fobj.read())
+        with smudge(scm.lfs_storage, sys.stdin.buffer) as fobj_:
+            sys.stdout.buffer.write(fobj_.read())
     finally:
         scm.close()