Add typing to git_util.py, metadata.py and config.py

.gitignore

@@ -4,6 +4,7 @@ __pycache__/
 .ipynb_checkpoints/
 .tox/
 dist/
+build/
 docs/_build/
 apps_meta.sqlite
 Pipfile.lock

aiidalab/config.py

@@ -1,6 +1,7 @@
 """Module to manage AiiDAlab configuration."""
 from os import getenv
 from pathlib import Path
+from typing import Any, Optional
 
 import click
 import toml
@@ -18,11 +19,11 @@
     click.secho("\n".join([f"\U0001F6A7  {line}" for line in lines]), fg="yellow")
 
 
-def _as_env_var_name(key):
+def _as_env_var_name(key: str) -> str:
     return "AIIDALAB_" + key.upper()
 
 
-def _get_config_value(key, default=None):
+def _get_config_value(key: str, default: Optional[str] = None) -> Any:
     """Return config value from configuration source.
 
     The standard configuration source order is:

aiidalab/git_util.py

@@ -1,11 +1,25 @@
 """Utility module for git-managed AiiDAlab apps."""
+
+# This future import turns on postponed evaluation of annotations, per PEP 563.
+# https://peps.python.org/pep-0563/
+# This is needed for two reasons:
+# 1. Using the new Union syntax (type1 | type2) with Python < 3.10
+# 2. Instead of using the Self type when returning an instance of the class,
+#    which is only available since 3.11, we can use the class directly,
+#    because the annotation evaluation is deffered (as per PEP) See:
+#    https://realpython.com/python-type-self/#using-the-__future__-module
+# For the 2., we could instead import Self from type_extensions, see:
+# https://realpython.com/python-type-self/#how-to-annotate-a-method-with-the-self-type-in-python
+from __future__ import annotations
+
 import locale
 import os
 import re
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
 from subprocess import CalledProcessError, run
+from typing import Any, Generator
 from urllib.parse import urldefrag
 
 from dulwich.porcelain import branch_list, status
@@ -21,14 +35,14 @@ class BranchTrackingStatus(Enum):
     DIVERGED = 2
 
 
-class GitManagedAppRepo(Repo):
+class GitManagedAppRepo(Repo):  # type: ignore
     """Utility class to simplify management of git-based apps."""
 
-    def list_branches(self):
+    def list_branches(self) -> Any:
         """List all repository branches."""
         return branch_list(self)
 
-    def branch(self):
+    def branch(self) -> bytes:
         """Return the current branch.
 
         Raises RuntimeError if the repository is in a detached HEAD state.
@@ -38,7 +52,7 @@ def branch(self):
             return branches[0]
         raise RuntimeError("In detached HEAD state.")
 
-    def get_tracked_branch(self, branch=None):
+    def get_tracked_branch(self, branch: bytes | None = None) -> Any:
         """Return the tracked branch for a given branch or None if the branch is not tracking."""
         if branch is None:
             branch = self.branch()
@@ -53,19 +67,19 @@ def get_tracked_branch(self, branch=None):
         except KeyError:
             return None
 
-    def dirty(self):
+    def dirty(self) -> bool:
         """Check if there are likely local user modifications to the app repository."""
         status_ = status(self)
         return bool(any(bool(_) for _ in status_.staged.values()) or status_.unstaged)
 
-    def update_available(self):
+    def update_available(self) -> bool:
         """Check whether there non-pulled commits on the tracked branch."""
         return (
             self.get_branch_tracking_status(self.branch())
             is BranchTrackingStatus.BEHIND
         )
 
-    def get_branch_tracking_status(self, branch):
+    def get_branch_tracking_status(self, branch: bytes) -> BranchTrackingStatus | None:
         """Return the tracking status of branch."""
         tracked_branch = self.get_tracked_branch(branch)
         if tracked_branch:
@@ -89,7 +103,7 @@ def get_branch_tracking_status(self, branch):
 
         return None
 
-    def _get_branch_for_ref(self, ref):
+    def _get_branch_for_ref(self, ref: bytes) -> list[bytes]:
         """Get the branch name for a given reference."""
         pattern = rb"refs\/heads\/"
         return [
@@ -99,7 +113,7 @@ def _get_branch_for_ref(self, ref):
         ]
 
 
-def git_clone(url, commit, path):
+def git_clone(url, commit, path: Path):  # type: ignore
     try:
         run(
             ["git", "clone", str(url), str(path)],
@@ -120,22 +134,22 @@ def git_clone(url, commit, path):
 
 
 @dataclass
-class GitPath(os.PathLike):
+class GitPath(os.PathLike):  # type: ignore
     """Utility class to operate on git objects like path objects."""
 
     repo: Path
     commit: str
     path: Path = Path(".")
 
-    def __fspath__(self):
+    def __fspath__(self) -> str:
         return str(Path(self.repo).joinpath(self.path))
 
-    def joinpath(self, *other):
+    def joinpath(self, *other: str) -> GitPath:
         return type(self)(
             repo=self.repo, path=self.path.joinpath(*other), commit=self.commit
         )
 
-    def resolve(self, strict=False):
+    def resolve(self, strict: bool = False) -> GitPath:
         return type(self)(
             repo=self.repo,
             path=Path(self.repo)
@@ -145,7 +159,7 @@ def resolve(self, strict=False):
             commit=self.commit,
         )
 
-    def _get_type(self):
+    def _get_type(self) -> str | None:
         # Git expects that a current directory path ("." or "./") is
         # represented with a trailing slash for this function.
         path = "./" if self.path == Path() else self.path
@@ -166,13 +180,13 @@ def _get_type(self):
                 return None
             raise
 
-    def is_file(self):
+    def is_file(self) -> bool:
         return self._get_type() == "blob"
 
-    def is_dir(self):
+    def is_dir(self) -> bool:
         return self._get_type() == "tree"
 
-    def read_bytes(self):
+    def read_bytes(self) -> bytes:
         try:
             return run(
                 ["git", "show", f"{self.commit}:{self.path}"],
@@ -195,16 +209,16 @@ def read_bytes(self):
             else:
                 raise  # unexpected error
 
-    def read_text(self, encoding=None, errors=None):
+    def read_text(self, encoding: str | None = None, errors: str | None = None) -> str:
         if encoding is None:
             encoding = locale.getpreferredencoding(False)
         if errors is None:
             errors = "strict"
         return self.read_bytes().decode(encoding=encoding, errors=errors)
 
 
-class GitRepo(Repo):
-    def get_current_branch(self):
+class GitRepo(Repo):  # type: ignore
+    def get_current_branch(self) -> str | None:
         try:
             branch = run(
                 ["git", "branch", "--show-current"],
@@ -223,10 +237,10 @@ def get_current_branch(self):
             )
         return branch.strip()
 
-    def get_commit_for_tag(self, tag):
+    def get_commit_for_tag(self, tag: str) -> Any:
         return self.get_peeled(f"refs/tags/{tag}".encode()).decode()
 
-    def get_merged_tags(self, branch):
+    def get_merged_tags(self, branch: str) -> Generator[str, None, None]:
         for branch_ref in [f"refs/heads/{branch}", f"refs/remotes/origin/{branch}"]:
             if branch_ref.encode() in self.refs:
                 yield from run(
@@ -240,7 +254,7 @@ def get_merged_tags(self, branch):
         else:
             raise ValueError(f"Not a branch: {branch}")
 
-    def rev_list(self, rev_selection):
+    def rev_list(self, rev_selection: str) -> list[str]:
         return run(
             ["git", "rev-list", rev_selection],
             cwd=self.path,
@@ -249,7 +263,7 @@ def rev_list(self, rev_selection):
             capture_output=True,
         ).stdout.splitlines()
 
-    def ref_from_rev(self, rev):
+    def ref_from_rev(self, rev: str) -> str:
         """Determine ref from rev.
 
         Returns branch reference if a branch with that name exists, otherwise a
@@ -266,7 +280,7 @@ def ref_from_rev(self, rev):
             return rev
 
     @classmethod
-    def clone_from_url(cls, url, path):
+    def clone_from_url(cls, url: str, path: str) -> GitRepo:
         run(
             ["git", "clone", urldefrag(url).url, path],
             cwd=Path(path).parent,

aiidalab/metadata.py

@@ -1,15 +1,17 @@
 """App metadata specification"""
 from __future__ import annotations
 
-from configparser import ConfigParser
+from configparser import ConfigParser, SectionProxy
 from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Generator
 
 __all__ = [
     "Metadata",
 ]
 
 
-def _map_development_state(classifiers):
+def _map_development_state(classifiers: str | list[str]) -> str:
     "Map standard trove classifiers (PEP 301) to aiidalab development states."
     if "Development Status :: 1 - Planning" in classifiers:
         return "registered"
@@ -28,24 +30,27 @@ def _map_development_state(classifiers):
         return "registered"
 
 
-def _parse_config_dict(dict_):
+def _parse_config_dict(dict_: str) -> Generator[tuple[str, str], None, None]:
     "Parse a config dict string for key-values pairs."
     for line in dict_.splitlines():
         if line:
             key, value = line.split("=")
             yield key.strip(), value.strip()
 
 
-def _parse_setup_cfg(setup_cfg):
+def _parse_setup_cfg(
+    setup_cfg: str,
+) -> Generator[tuple[str, str | list[str]], None, None]:
     "Parse a setup.cfg configuration file string for metadata."
     cfg = ConfigParser()
     cfg.read_string(setup_cfg)
 
-    try:
-        metadata_pep426 = cfg["metadata"]
-    except KeyError:
-        metadata_pep426 = dict()
-    aiidalab = cfg["aiidalab"] if "aiidalab" in cfg else dict()
+    metadata_pep426: SectionProxy | dict[Any, Any] = (
+        cfg["metadata"] if "metadata" in cfg else {}
+    )
+    aiidalab: SectionProxy | dict[Any, Any] = (
+        cfg["aiidalab"] if "aiidalab" in cfg else {}
+    )
 
     yield "title", aiidalab.get("title", metadata_pep426.get("name"))
     yield "version", aiidalab.get("version", metadata_pep426.get("version"))
@@ -62,12 +67,12 @@ def _parse_setup_cfg(setup_cfg):
         "logo", project_urls.get("Logo") or project_urls.get("logo")
     )
     yield "state", aiidalab.get(
-        "state", _map_development_state(metadata_pep426.get("classifiers", []))
+        "state", _map_development_state(metadata_pep426.get("classifiers", ""))
     )
 
     # Allow passing single category and convert to list
     # and allow parse line separated string as list
-    categories = aiidalab.get("categories", [])
+    categories = aiidalab.get("categories", "")
     if isinstance(categories, str):
         categories = [c for c in categories.split("\n") if c]
     yield "categories", categories
@@ -90,7 +95,7 @@ class Metadata:
     _search_dirs = (".aiidalab", "./")
 
     @staticmethod
-    def _parse(path):
+    def _parse(path: Path) -> dict[str, Any]:
         try:
             return {
                 key: value
@@ -103,7 +108,7 @@ def _parse(path):
             return {}
 
     @classmethod
-    def parse(cls, root):
+    def parse(cls, root: Path) -> Metadata:
         """Parse the app metadata from a setup.cfg within the app repository.
 
         This function will parse metadata fields from a possible "aiidalab"

setup.cfg

@@ -82,21 +82,12 @@ disallow_untyped_calls = False
 disallow_incomplete_defs = True
 warn_return_any = True
 
-[mypy-aiidalab.metadata.*]
-ignore_errors = True
-
-[mypy-aiidalab.git_util.*]
-ignore_errors = True
-
 [mypy-aiidalab.__main__.*]
 ignore_errors = True
 
 [mypy-aiidalab.registry.*]
 ignore_errors = True
 
-[mypy-aiidalab.config.*]
-ignore_errors = True
-
 [mypy-aiidalab.fetch.*]
 ignore_errors = True