refactor!: Add typings

Author: tony

conftest.py

@@ -34,36 +34,40 @@ def cwd_default(monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path) -> None
 
 @pytest.fixture(autouse=True, scope="session")
 @pytest.mark.usefixtures("set_home")
-def xdg_config_path(user_path: pathlib.Path):
+def xdg_config_path(user_path: pathlib.Path) -> pathlib.Path:
     p = user_path / ".config"
     p.mkdir()
     return p
 
 
 @pytest.fixture(scope="function")
-def config_path(xdg_config_path: pathlib.Path, request: pytest.FixtureRequest):
+def config_path(
+    xdg_config_path: pathlib.Path, request: pytest.FixtureRequest
+) -> pathlib.Path:
     conf_path = xdg_config_path / "vcspull"
     conf_path.mkdir(exist_ok=True)
 
-    def clean():
+    def clean() -> None:
         shutil.rmtree(conf_path)
 
     request.addfinalizer(clean)
     return conf_path
 
 
 @pytest.fixture(autouse=True)
-def set_xdg_config_path(monkeypatch: pytest.MonkeyPatch, xdg_config_path: pathlib.Path):
+def set_xdg_config_path(
+    monkeypatch: pytest.MonkeyPatch, xdg_config_path: pathlib.Path
+) -> None:
     monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg_config_path))
 
 
 @pytest.fixture(scope="function")
-def repos_path(user_path: pathlib.Path, request: pytest.FixtureRequest):
+def repos_path(user_path: pathlib.Path, request: pytest.FixtureRequest) -> pathlib.Path:
     """Return temporary directory for repository checkout guaranteed unique."""
     dir = user_path / "repos"
     dir.mkdir(exist_ok=True)
 
-    def clean():
+    def clean() -> None:
         shutil.rmtree(dir)
 
     request.addfinalizer(clean)

docs/conf.py

@@ -1,8 +1,13 @@
 # flake8: noqa E501
+import inspect
 import os
 import sys
+import typing as t
+from os.path import dirname, relpath
 from pathlib import Path
 
+import vcspull
+
 # Get the project root dir, which is the parent dir of this
 cwd = Path.cwd()
 project_root = cwd.parent
@@ -60,7 +65,7 @@
 html_css_files = ["css/custom.css"]
 html_extra_path = ["manifest.json"]
 html_theme = "furo"
-html_theme_path: list = []
+html_theme_path: list[str] = []
 html_theme_options = {
     "light_logo": "img/vcspull.svg",
     "dark_logo": "img/vcspull-dark.svg",
@@ -171,13 +176,9 @@
 }
 
 
-def linkcode_resolve(domain, info):  # NOQA: C901
-    import inspect
-    import sys
-    from os.path import dirname, relpath
-
-    import vcspull
-
+def linkcode_resolve(
+    domain: str, info: dict[str, str]
+) -> t.Union[None, str]:  # NOQA: C901
     """
     Determine the URL corresponding to Python object
 
@@ -210,7 +211,8 @@ def linkcode_resolve(domain, info):  # NOQA: C901
     except AttributeError:
         pass
     else:
-        obj = unwrap(obj)
+        if callable(obj):
+            obj = unwrap(obj)
 
     try:
         fn = inspect.getsourcefile(obj)

scripts/generate_gitlab.py

@@ -7,6 +7,10 @@
 import requests
 import yaml
 
+from libvcs.sync.git import GitRemote
+from vcspull.cli.sync import guess_vcs
+from vcspull.types import RawConfig
+
 try:
     gitlab_token = os.environ["GITLAB_TOKEN"]
 except KeyError:
@@ -69,24 +73,37 @@
     sys.exit(1)
 
 path_prefix = os.getcwd()
-config: dict = {}
+config: RawConfig = {}
 
 for group in response.json():
     url_to_repo = group["ssh_url_to_repo"].replace(":", "/")
     namespace_path = group["namespace"]["full_path"]
     reponame = group["path"]
 
-    path = "%s/%s" % (path_prefix, namespace_path)
+    path = f"{path_prefix}/{namespace_path}"
 
     if path not in config:
         config[path] = {}
 
     # simplified config not working - https://github.com/vcs-python/vcspull/issues/332
     # config[path][reponame] = 'git+ssh://%s' % (url_to_repo)
 
+    vcs = guess_vcs(url_to_repo)
+    if vcs is None:
+        raise Exception(f"Could not guess VCS for URL: {url_to_repo}")
+
     config[path][reponame] = {
-        "url": "git+ssh://%s" % (url_to_repo),
-        "remotes": {"origin": "ssh://%s" % (url_to_repo)},
+        "name": reponame,
+        "dir": path / reponame,
+        "url": f"git+ssh://{url_to_repo}",
+        "remotes": {
+            "origin": GitRemote(
+                name="origin",
+                fetch_url=f"ssh://{url_to_repo}",
+                push_url=f"ssh://{url_to_repo}",
+            )
+        },
+        "vcs": vcs,
     }
 
 config_yaml = yaml.dump(config)

src/vcspull/_internal/config_reader.py

@@ -22,11 +22,11 @@ class ConfigReader:
     '{\n  "session_name": "my session"\n}'
     """
 
-    def __init__(self, content: "RawConfigData"):
+    def __init__(self, content: "RawConfigData") -> None:
         self.content = content
 
     @staticmethod
-    def _load(format: "FormatLiteral", content: str):
+    def _load(format: "FormatLiteral", content: str) -> t.Dict[str, t.Any]:
         """Load raw config data and directly return it.
 
         >>> ConfigReader._load("json", '{ "session_name": "my session" }')
@@ -36,17 +36,20 @@ def _load(format: "FormatLiteral", content: str):
         {'session_name': 'my session'}
         """
         if format == "yaml":
-            return yaml.load(
-                content,
-                Loader=yaml.SafeLoader,
+            return t.cast(
+                t.Dict[str, t.Any],
+                yaml.load(
+                    content,
+                    Loader=yaml.SafeLoader,
+                ),
             )
         elif format == "json":
-            return json.loads(content)
+            return t.cast(t.Dict[str, t.Any], json.loads(content))
         else:
             raise NotImplementedError(f"{format} not supported in configuration")
 
     @classmethod
-    def load(cls, format: "FormatLiteral", content: str):
+    def load(cls, format: "FormatLiteral", content: str) -> "ConfigReader":
         """Load raw config data into a ConfigReader instance (to dump later).
 
         >>> cfg = ConfigReader.load("json", '{ "session_name": "my session" }')
@@ -69,7 +72,7 @@ def load(cls, format: "FormatLiteral", content: str):
         )
 
     @classmethod
-    def _from_file(cls, path: pathlib.Path):
+    def _from_file(cls, path: pathlib.Path) -> t.Dict[str, t.Any]:
         r"""Load data from file path directly to dictionary.
 
         **YAML file**
@@ -102,7 +105,7 @@ def _from_file(cls, path: pathlib.Path):
         content = open(path).read()
 
         if path.suffix in [".yaml", ".yml"]:
-            format: FormatLiteral = "yaml"
+            format: "FormatLiteral" = "yaml"
         elif path.suffix == ".json":
             format = "json"
         else:
@@ -114,7 +117,7 @@ def _from_file(cls, path: pathlib.Path):
         )
 
     @classmethod
-    def from_file(cls, path: pathlib.Path):
+    def from_file(cls, path: pathlib.Path) -> "ConfigReader":
         r"""Load data from file path
 
         **YAML file**
@@ -169,11 +172,14 @@ def _dump(
         '{\n  "session_name": "my session"\n}'
         """
         if format == "yaml":
-            return yaml.dump(
-                content,
-                indent=2,
-                default_flow_style=False,
-                Dumper=yaml.SafeDumper,
+            return t.cast(
+                str,
+                yaml.dump(
+                    content,
+                    indent=2,
+                    default_flow_style=False,
+                    Dumper=yaml.SafeDumper,
+                ),
             )
         elif format == "json":
             return json.dumps(

src/vcspull/cli/__init__.py

@@ -7,6 +7,8 @@
 import argparse
 import logging
 import textwrap
+import typing as t
+from typing import overload
 
 from libvcs.__about__ import __version__ as libvcs_version
 
@@ -30,7 +32,21 @@
 ).strip()
 
 
-def create_parser(return_subparsers: bool = False):
+@overload
+def create_parser(
+    return_subparsers: t.Literal[True],
+) -> t.Tuple[argparse.ArgumentParser, t.Any]:
+    ...
+
+
+@overload
+def create_parser(return_subparsers: t.Literal[False]) -> argparse.ArgumentParser:
+    ...
+
+
+def create_parser(
+    return_subparsers: bool = False,
+) -> t.Union[argparse.ArgumentParser, t.Tuple[argparse.ArgumentParser, t.Any]]:
     parser = argparse.ArgumentParser(
         prog="vcspull",
         formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -64,9 +80,9 @@ def create_parser(return_subparsers: bool = False):
     return parser
 
 
-def cli(args=None):
+def cli(_args: t.Optional[t.List[str]] = None) -> None:
     parser, sync_parser = create_parser(return_subparsers=True)
-    args = parser.parse_args(args)
+    args = parser.parse_args(_args)
 
     setup_logger(log=log, level=args.log_level.upper())
 

src/vcspull/cli/sync.py

@@ -1,18 +1,22 @@
 import argparse
 import logging
+import pathlib
 import sys
 import typing as t
 from copy import deepcopy
+from datetime import datetime
 
 from libvcs._internal.shortcuts import create_project
+from libvcs._internal.types import VCSLiteral
+from libvcs.sync.git import GitSync
 from libvcs.url import registry as url_tools
 
 from ..config import filter_repos, find_config_files, load_configs
 
 log = logging.getLogger(__name__)
 
 
-def clamp(n, _min, _max):
+def clamp(n: int, _min: int, _max: int) -> int:
     return max(_min, min(n, _max))
 
 
@@ -51,8 +55,8 @@ def create_sync_subparser(parser: argparse.ArgumentParser) -> argparse.ArgumentP
 
 
 def sync(
-    repo_patterns,
-    config,
+    repo_patterns: t.List[str],
+    config: pathlib.Path,
     exit_on_error: bool,
     parser: t.Optional[
         argparse.ArgumentParser
@@ -102,12 +106,28 @@ def sync(
                     raise SystemExit(EXIT_ON_ERROR_MSG)
 
 
-def progress_cb(output, timestamp):
+def progress_cb(output: str, timestamp: datetime) -> None:
     sys.stdout.write(output)
     sys.stdout.flush()
 
 
-def update_repo(repo_dict):
+def guess_vcs(url: str) -> t.Optional[VCSLiteral]:
+    vcs_matches = url_tools.registry.match(url=url, is_explicit=True)
+
+    if len(vcs_matches) == 0:
+        log.warning(f"No vcs found for {url}")
+        return None
+    if len(vcs_matches) > 1:
+        log.warning(f"No exact matches for {url}")
+        return None
+
+    return t.cast(VCSLiteral, vcs_matches[0].vcs)
+
+
+def update_repo(
+    repo_dict: t.Any,
+    # repo_dict: Dict[str, Union[str, Dict[str, GitRemote], pathlib.Path]]
+) -> GitSync:
     repo_dict = deepcopy(repo_dict)
     if "pip_url" not in repo_dict:
         repo_dict["pip_url"] = repo_dict.pop("url")
@@ -116,16 +136,16 @@ def update_repo(repo_dict):
     repo_dict["progress_callback"] = progress_cb
 
     if repo_dict.get("vcs") is None:
-        vcs_matches = url_tools.registry.match(url=repo_dict["url"], is_explicit=True)
-
-        if len(vcs_matches) == 0:
-            raise Exception(f"No vcs found for {repo_dict}")
-        if len(vcs_matches) > 1:
-            raise Exception(f"No exact matches for {repo_dict}")
+        vcs = guess_vcs(url=repo_dict["url"])
+        if vcs is None:
+            raise Exception(
+                f'Could not automatically determine VCS for {repo_dict["url"]}'
+            )
 
-        repo_dict["vcs"] = vcs_matches[0].vcs
+        repo_dict["vcs"] = vcs
 
     r = create_project(**repo_dict)  # Creates the repo object
     r.update_repo(set_remotes=True)  # Creates repo if not exists and fetches
 
-    return r
+    # TODO: Fix this
+    return r  # type:ignore