Add types with enforcement to SBOM and buildbotapi (#151)

Author: ambv

buildbotapi.py

@@ -1,20 +1,22 @@
 import asyncio
 import json
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, cast
 
 from aiohttp.client import ClientSession
 
+JSON = dict[str, Any]
+
 
 @dataclass
 class Builder:
     builderid: int
-    description: Optional[str]
-    masterids: List[int]
+    description: str | None
+    masterids: list[int]
     name: str
-    tags: List[str]
+    tags: list[str]
 
-    def __init__(self, **kwargs) -> None:
+    def __init__(self, **kwargs: Any) -> None:
         self.__dict__.update(**kwargs)
 
     def __hash__(self) -> int:
@@ -25,15 +27,17 @@ def __hash__(self) -> int:
 class Build:
     id: int
     is_currently_failing: bool
+    builderid: int
+    builder: Builder | None
 
-    def __init__(self, **kwargs):
+    def __init__(self, **kwargs: Any) -> None:
         self.__dict__.update(**kwargs)
-        self.id = kwargs.get("number")
-        self.is_currently_failing = kwargs.get("currently_failing")
+        self.id = int(kwargs.get("number", -1))
+        self.is_currently_failing = kwargs.get("currently_failing", False)
         self.builder = None
 
-    def __eq__(self, other):
-        return self.id == other.id
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, Build) and self.id == other.id
 
     def __hash__(self) -> int:
         return hash(self.id)
@@ -52,37 +56,30 @@ async def _fetch_text(self, url: str) -> str:
         async with self._session.get(url) as resp:
             return await resp.text()
 
-    async def _fetch_json(self, url: str) -> Dict[
-        str,
-        Union[
-            List[Dict[str, Union[int, bool, str]]],
-            Dict[str, int],
-            List[Dict[str, Optional[Union[int, List[int], str, List[str]]]]],
-        ],
-    ]:
-        return json.loads(await self._fetch_text(url))
-
-    async def stable_builders(self, branch: Optional[str]) -> Dict[int, Builder]:
+    async def _fetch_json(self, url: str) -> JSON:
+        return cast(JSON, json.loads(await self._fetch_text(url)))
+
+    async def stable_builders(self, branch: str | None = None) -> dict[int, Builder]:
         stable_builders = {
             id: builder
             for (id, builder) in (await self.all_builders(branch=branch)).items()
             if "stable" in builder.tags
         }
         return stable_builders
 
-    async def all_builders(self, branch: Optional[str] = None) -> Dict[int, Builder]:
+    async def all_builders(self, branch: str | None = None) -> dict[int, Builder]:
         url = "https://buildbot.python.org/all/api/v2/builders"
         if branch is not None:
             url = f"{url}?tags__contains={branch}"
-        _builders: Dict[str, Any] = await self._fetch_json(url)
+        _builders: dict[str, Any] = await self._fetch_json(url)
         builders = _builders["builders"]
         all_builders = {
             builder["builderid"]: Builder(**builder) for builder in builders
         }
         return all_builders
 
     async def is_builder_failing_currently(self, builder: Builder) -> bool:
-        builds_: Dict[str, Any] = await self._fetch_json(
+        builds_: dict[str, Any] = await self._fetch_json(
             f"https://buildbot.python.org/all/api/v2/builds?complete__eq=true"
             f"&&builderid__eq={builder.builderid}&&order=-complete_at"
             f"&&limit=1"
@@ -95,13 +92,13 @@ async def is_builder_failing_currently(self, builder: Builder) -> bool:
             return True
         return False
 
-    async def get_build(self, builder_id, build_id):
+    async def get_build(self, builder_id: int, build_id: int) -> Build:
         data = await self._fetch_json(
             f"https://buildbot.python.org/all/api/v2/builders/{builder_id}"
             f"/builds/{build_id}"
         )
         (build_data,) = data["builds"]
-        build = Build(**build_data)
+        build: Build = Build(**build_data)
         build.builder = (await self.all_builders())[build.builderid]
         build.is_currently_failing = await self.is_builder_failing_currently(
             build.builder
@@ -126,7 +123,8 @@ async def get_recent_failures(self, limit: int = 100) -> set[Build]:
         for failure in all_failures:
             failure.builder = stable_builders[failure.builderid]
 
-        async def _get_missing_info(failure):
+        async def _get_missing_info(failure: Build) -> None:
+            assert failure.builder is not None
             failure.is_currently_failing = await self.is_builder_failing_currently(
                 failure.builder
             )

mypy-requirements.txt

@@ -4,3 +4,4 @@ pytest
 pytest-mock
 sigstore==1.1.2
 types-requests
+types-paramiko

pyproject.toml

@@ -9,9 +9,7 @@ extra_checks = true
 warn_unreachable = true
 
 exclude = [
-  "^buildbotapi.py$",
   "^run_release.py$",
-  "^sbom.py$",
   "^tests/test_release_tag.py$",
   "^tests/test_run_release.py$",
   "^tests/test_sbom.py$",