Add type hints to run_release.py (#137)

Author: hugovk

buildbotapi.py

@@ -35,7 +35,7 @@ def __init__(self, **kwargs):
     def __eq__(self, other):
         return self.id == other.id
 
-    def __hash__(self):
+    def __hash__(self) -> int:
         return hash(self.id)
 
 
@@ -108,7 +108,7 @@ async def get_build(self, builder_id, build_id):
         )
         return build
 
-    async def get_recent_failures(self, limit=100):
+    async def get_recent_failures(self, limit: int = 100) -> set[Build]:
         data = await self._fetch_json(
             f"https://buildbot.python.org/all/api/v2/builds?"
             f"complete__eq=true&&results__eq=2&&"

mypy-requirements.txt

@@ -1,3 +1,5 @@
+aiohttp==3.9.4
 mypy==1.10
 pytest
 pytest-mock
+sigstore==1.1.2

run_release.py

@@ -4,6 +4,7 @@
 
 Original code by Pablo Galindo
 """
+from __future__ import annotations
 
 import argparse
 import asyncio
@@ -23,7 +24,7 @@
 import urllib.request
 from dataclasses import dataclass
 from shelve import DbfilenameShelf
-from typing import Callable, Iterator, List, Optional
+from typing import Any, Callable, Generator, Iterator
 
 import aiohttp
 import gnupg
@@ -33,7 +34,7 @@
 
 import release as release_mod
 import sbom
-from buildbotapi import BuildBotAPI
+from buildbotapi import BuildBotAPI, Builder
 
 API_KEY_REGEXP = re.compile(r"(?P<major>\w+):(?P<minor>\w+)")
 
@@ -184,7 +185,7 @@ class Task:
     function: Callable[[DbfilenameShelf], None]
     description: str
 
-    def __call__(self, db: DbfilenameShelf) -> None:
+    def __call__(self, db: DbfilenameShelf) -> Any:
         return getattr(self, "function")(db)
 
 
@@ -195,13 +196,13 @@ class ReleaseException(Exception):
 class ReleaseDriver:
     def __init__(
         self,
-        tasks: List[Task],
+        tasks: list[Task],
         *,
         release_tag: release_mod.Tag,
         git_repo: str,
         api_key: str,
         ssh_user: str,
-        first_state: Optional[Task] = None,
+        first_state: Task | None = None,
     ) -> None:
         self.tasks = tasks
         dbfile = pathlib.Path.home() / ".python_release"
@@ -212,7 +213,7 @@ def __init__(
             self.db.close()
             self.db = shelve.open(str(dbfile), "n")
 
-        self.current_task: Optional[Task] = first_state
+        self.current_task: Task | None = first_state
         self.completed_tasks = self.db.get("completed_tasks", [])
         self.remaining_tasks = iter(tasks[len(self.completed_tasks) :])
         if self.db.get("gpg_key"):
@@ -282,7 +283,7 @@ def cd(path: str) -> Iterator[None]:
 
 
 @contextlib.contextmanager
-def supress_print():
+def supress_print() -> Generator[None, None, None]:
     print_func = builtins.print
     builtins.print = lambda *args, **kwargs: None
     yield
@@ -333,8 +334,10 @@ def check_ssh_connection(db: DbfilenameShelf) -> None:
 
 
 def check_buildbots(db: DbfilenameShelf) -> None:
-    async def _check():
-        async def _get_builder_status(buildbot_api, the_builder):
+    async def _check() -> set[Builder]:
+        async def _get_builder_status(
+            buildbot_api: BuildBotAPI, the_builder: Builder
+        ) -> tuple[Builder, bool]:
             return the_builder, await buildbot_api.is_builder_failing_currently(
                 the_builder
             )
@@ -567,7 +570,7 @@ def build_sbom_artifacts(db):
 
 
 class MySFTPClient(paramiko.SFTPClient):
-    def put_dir(self, source, target, progress=None):
+    def put_dir(self, source: str, target: str, progress: Any = None) -> None:
         for item in os.listdir(source):
             if os.path.isfile(os.path.join(source, item)):
                 progress.text(item)
@@ -581,7 +584,7 @@ def put_dir(self, source, target, progress=None):
                     progress=progress,
                 )
 
-    def mkdir(self, path, mode=511, ignore_existing=False):
+    def mkdir(self, path: str, mode: int = 511, ignore_existing: bool = False) -> None:
         try:
             super().mkdir(path, mode)
         except OSError:
@@ -609,7 +612,7 @@ def upload_files_to_server(db: DbfilenameShelf) -> None:
 
     shutil.rmtree(artifacts_path / f"Python-{db['release']}", ignore_errors=True)
 
-    def upload_subdir(subdir):
+    def upload_subdir(subdir: str) -> None:
         with contextlib.suppress(OSError):
             ftp_client.mkdir(str(destination / subdir))
         with alive_bar(len(tuple((artifacts_path / subdir).glob("**/*")))) as progress:
@@ -636,7 +639,7 @@ def place_files_in_download_folder(db: DbfilenameShelf) -> None:
     source = f"/home/psf-users/{db['ssh_user']}/{db['release']}"
     destination = f"/srv/www.python.org/ftp/python/{db['release'].normalized()}"
 
-    def execute_command(command):
+    def execute_command(command: str) -> None:
         channel = client.get_transport().open_session()
         channel.exec_command(command)
         if channel.recv_exit_status() != 0:
@@ -655,7 +658,7 @@ def execute_command(command):
         source = f"/home/psf-users/{db['ssh_user']}/{db['release']}"
         destination = f"/srv/www.python.org/ftp/python/doc/{release_tag}"
 
-        def execute_command(command):
+        def execute_command(command: str) -> None:
             channel = client.get_transport().open_session()
             channel.exec_command(command)
             if channel.recv_exit_status() != 0:
@@ -690,7 +693,7 @@ def upload_docs_to_the_docs_server(db: DbfilenameShelf) -> None:
 
     shutil.rmtree(artifacts_path / f"Python-{db['release']}", ignore_errors=True)
 
-    def upload_subdir(subdir):
+    def upload_subdir(subdir: str) -> None:
         with contextlib.suppress(OSError):
             ftp_client.mkdir(str(destination / subdir))
         with alive_bar(len(tuple((artifacts_path / subdir).glob("**/*")))) as progress:
@@ -719,7 +722,7 @@ def unpack_docs_in_the_docs_server(db: DbfilenameShelf) -> None:
     source = f"/home/psf-users/{db['ssh_user']}/{db['release']}"
     destination = f"/srv/docs.python.org/release/{release_tag}"
 
-    def execute_command(command):
+    def execute_command(command: str) -> None:
         channel = client.get_transport().open_session()
         channel.exec_command(command)
         if channel.recv_exit_status() != 0:
@@ -926,8 +929,6 @@ def modify_the_release_to_the_prerelease_pages(db: DbfilenameShelf) -> None:
 
 
 def post_release_merge(db: DbfilenameShelf) -> None:
-    release_tag: release_mod.Tag = db["release"]
-
     subprocess.check_call(
         ["git", "fetch", "--all"],
         cwd=db["git_repo"],
@@ -1031,7 +1032,7 @@ def is_mirror(repo: pathlib.Path, remote: str) -> bool:
 
 
 def push_to_local_fork(db: DbfilenameShelf) -> None:
-    def _push_to_local(dry_run=False):
+    def _push_to_local(dry_run: bool = False) -> None:
         git_command = ["git", "push"]
         if dry_run:
             git_command.append("--dry-run")
@@ -1057,7 +1058,7 @@ def _push_to_local(dry_run=False):
 def push_to_upstream(db: DbfilenameShelf) -> None:
     release_tag: release_mod.Tag = db["release"]
 
-    def _push_to_upstream(dry_run=False):
+    def _push_to_upstream(dry_run: bool = False) -> None:
         branch = f"{release_tag.major}.{release_tag.minor}"
         git_command = ["git", "push"]
         if dry_run:
@@ -1112,7 +1113,7 @@ def main() -> None:
 
     parser = argparse.ArgumentParser(description="Process some integers.")
 
-    def _release_type(release):
+    def _release_type(release: str) -> str:
         if not RELEASE_REGEXP.match(release):
             raise argparse.ArgumentTypeError("Invalid release string")
         return release
@@ -1132,7 +1133,7 @@ def _release_type(release):
         type=str,
     )
 
-    def _api_key(api_key):
+    def _api_key(api_key: str) -> str:
         if not API_KEY_REGEXP.match(api_key):
             raise argparse.ArgumentTypeError(
                 "Invalid api key format. It must be on the form USER:API_KEY"

sbom.py

@@ -23,6 +23,7 @@
 import tarfile
 import typing
 import zipfile
+from typing import Any
 from urllib.request import urlopen
 
 
@@ -118,13 +119,13 @@ def get_release_tools_commit_sha() -> str:
     return stdout
 
 
-def normalize_sbom_data(sbom_data):
+def normalize_sbom_data(sbom_data: dict[str, Any]) -> None:
     """
     Normalize SBOM data in-place by recursion
     and sorting lists by some repeatable key.
     """
 
-    def recursive_sort_in_place(value):
+    def recursive_sort_in_place(value: list | dict) -> None:
         if isinstance(value, list):
             # We need to recurse first so bottom-most elements are sorted first.
             for item in value:
@@ -338,7 +339,7 @@ def create_cpython_sbom(
     sbom_data: dict[str, typing.Any],
     cpython_version: str,
     artifact_path: str,
-):
+) -> None:
     """Creates the top-level SBOM metadata and the CPython SBOM package."""
 
     cpython_version_without_suffix = re.match(r"^([0-9.]+)", cpython_version).group(1)
@@ -412,7 +413,7 @@ def create_cpython_sbom(
     sbom_data["packages"].append(sbom_cpython_package)
 
 
-def create_sbom_for_source_tarball(tarball_path: str):
+def create_sbom_for_source_tarball(tarball_path: str) -> dict[str, Any]:
     """Stitches together an SBOM for a source tarball"""
     tarball_name = os.path.basename(tarball_path)
 
@@ -567,7 +568,9 @@ def create_sbom_for_source_tarball(tarball_path: str):
     return sbom_data
 
 
-def create_sbom_for_windows_artifact(artifact_path, cpython_source_dir: str):
+def create_sbom_for_windows_artifact(
+    artifact_path: str, cpython_source_dir: str
+) -> dict[str, Any]:
     artifact_name = os.path.basename(artifact_path)
     cpython_version = re.match(
         r"^python-([0-9abrc.]+)(?:-|\.exe|\.zip)", artifact_name