fix: type-check jq_collector under mypy strict (#52)

collector/jq_collector/github.py

@@ -112,7 +112,8 @@ def _json(self, path: str, **params: str | int) -> object | None:
             log.info("%s -> %s", path, response.status_code)
             return None
         response.raise_for_status()
-        return response.json()
+        payload: object = response.json()
+        return payload
 
     def _bytes(self, path: str) -> bytes | None:
         """GET returning raw bytes. 410 is added to the expected empties: that
@@ -125,8 +126,8 @@ def _bytes(self, path: str) -> bytes | None:
         response.raise_for_status()
         return response.content
 
-    def _paginate(self, path: str, **params: str | int) -> list[dict]:
-        items: list[dict] = []
+    def _paginate(self, path: str, **params: str | int) -> list[dict[str, Any]]:
+        items: list[dict[str, Any]] = []
         page = 1
         while True:
             batch = self._json(path, per_page=100, page=page, **params)
@@ -142,7 +143,7 @@ def _paginate(self, path: str, **params: str | int) -> list[dict]:
 
     # -- fleet-level -----------------------------------------------------
 
-    def list_repos(self, fleet: tuple[str, ...] | None = None) -> list[dict]:
+    def list_repos(self, fleet: tuple[str, ...] | None = None) -> list[dict[str, Any]]:
         """The repos named in the config, in the order they were listed.
 
         One call each, and no org sweep: the fleet is whatever you wrote down.
@@ -153,7 +154,7 @@ def list_repos(self, fleet: tuple[str, ...] | None = None) -> list[dict]:
         GitLab repo in a mixed fleet would be asked of GitHub, which answers 404
         and logs each one as unreadable.
         """
-        repos: list[dict] = []
+        repos: list[dict[str, Any]] = []
         seen: set[str] = set()
 
         for full_name in fleet if fleet is not None else self._cfg.repos:
@@ -171,7 +172,7 @@ def list_repos(self, fleet: tuple[str, ...] | None = None) -> list[dict]:
 
         return repos
 
-    def branch_protection(self, full_name: str, branch: str) -> tuple[dict | None, bool]:
+    def branch_protection(self, full_name: str, branch: str) -> tuple[dict[str, Any] | None, bool]:
         """``(protection, known)`` for one branch.
 
         The endpoint 404s both for an unprotected branch and for a token
@@ -284,7 +285,7 @@ def active_workflows(self, full_name: str) -> dict[int, str] | None:
                 seen[name] = wid
         return active
 
-    def latest_runs(self, full_name: str, branch: str) -> list[dict]:
+    def latest_runs(self, full_name: str, branch: str) -> list[dict[str, Any]]:
         """The newest completed run of each active workflow on ``branch``.
 
         The runs feed is ordered by ``created_at`` and is not a per-workflow
@@ -326,7 +327,7 @@ def latest_runs(self, full_name: str, branch: str) -> list[dict]:
             for wid, run in newest.items()
         ]
 
-    def _newest_conclusive(self, full_name: str, branch: str, wid: int) -> dict | None:
+    def _newest_conclusive(self, full_name: str, branch: str, wid: int) -> dict[str, Any] | None:
         """The newest run of one workflow that reached a verdict, if any."""
         extra = self._json(
             f"/repos/{full_name}/actions/workflows/{wid}/runs",
@@ -491,12 +492,14 @@ def _coverage(blob: bytes) -> tuple[float, int] | None:
     return round(float(rate) * 100, 1), int(root.get("lines-valid") or 0)
 
 
-def _newest_per_workflow(feed: list[dict], active: dict[int, str] | None) -> dict[Any, dict]:
+def _newest_per_workflow(
+    feed: list[dict[str, Any]], active: dict[int, str] | None
+) -> dict[Any, dict[str, Any]]:
     """The runs feed reduced to the newest conclusive run per active workflow.
 
     Keyed on workflow id, or on the run's name for a run that carries none.
     """
-    newest: dict[Any, dict] = {}
+    newest: dict[Any, dict[str, Any]] = {}
     for run in feed:
         wid = run.get("workflow_id")
         if active is not None and wid not in active:
@@ -510,7 +513,7 @@ def _newest_per_workflow(feed: list[dict], active: dict[int, str] | None) -> dic
     return newest
 
 
-def _inconclusive(run: dict) -> bool:
+def _inconclusive(run: dict[str, Any]) -> bool:
     return (run.get("conclusion") or "") in INCONCLUSIVE_CONCLUSIONS
 
 
@@ -581,7 +584,7 @@ def collect(
     )
     repos = [r for r in listing if r["full_name"] not in excluded]
 
-    def one(raw: dict) -> RemoteRepo:
+    def one(raw: dict[str, Any]) -> RemoteRepo:
         full_name = raw["full_name"]
         owner = (raw.get("owner") or {}).get("login") or full_name.split("/", 1)[0]
         branch = raw.get("default_branch") or "main"

collector/jq_collector/gitlab.py

@@ -40,6 +40,7 @@
 import concurrent.futures
 import logging
 from datetime import datetime
+from typing import Any
 from urllib.parse import quote
 
 import httpx
@@ -107,7 +108,8 @@ def _json(self, path: str, **params: str | int) -> object | None:
         response = self._get(path, **params)
         if response is None:
             return None
-        return response.json()
+        payload: object = response.json()
+        return payload
 
     def _get(self, path: str, **params: str | int) -> httpx.Response | None:
         """GET, or None for the statuses a real fleet legitimately produces."""
@@ -127,8 +129,8 @@ def _text(self, path: str, **params: str | int) -> str | None:
         response = self._get(path, **params)
         return response.text if response is not None else None
 
-    def _paginate(self, path: str, **params: str | int) -> list[dict]:
-        items: list[dict] = []
+    def _paginate(self, path: str, **params: str | int) -> list[dict[str, Any]]:
+        items: list[dict[str, Any]] = []
         page = 1
         while True:
             batch = self._json(path, per_page=_PER_PAGE, page=page, **params)
@@ -144,14 +146,14 @@ def _paginate(self, path: str, **params: str | int) -> list[dict]:
 
     # -- fleet-level -----------------------------------------------------
 
-    def list_projects(self, fleet: tuple[str, ...]) -> list[dict]:
+    def list_projects(self, fleet: tuple[str, ...]) -> list[dict[str, Any]]:
         """The projects named in the config, in the order they were listed.
 
         One call each, no group sweep - the same contract as
         ``github.GitHub.list_repos``, and for the same reason: the board's
         contents are decided by repos.yml and by nothing else.
         """
-        projects: list[dict] = []
+        projects: list[dict[str, Any]] = []
         seen: set[str] = set()
 
         for full_name in fleet:
@@ -170,7 +172,7 @@ def list_projects(self, fleet: tuple[str, ...]) -> list[dict]:
 
         return projects
 
-    def protected_branch(self, full_name: str, branch: str) -> dict | None:
+    def protected_branch(self, full_name: str, branch: str) -> dict[str, Any] | None:
         """The branch's protection entry, or None if it is not protected.
 
         Unlike GitHub's, this endpoint needs no admin rights, so there is no
@@ -208,7 +210,7 @@ def template_ref(self, full_name: str, branch: str) -> str:
             return ""
         return str(data.get("ref") or "") if isinstance(data, dict) else ""
 
-    def latest_pipeline(self, full_name: str, branch: str) -> dict | None:
+    def latest_pipeline(self, full_name: str, branch: str) -> dict[str, Any] | None:
         """The newest pipeline on *branch*, with its coverage.
 
         The listing carries a status but not coverage, so the one worth having
@@ -229,7 +231,7 @@ def latest_pipeline(self, full_name: str, branch: str) -> dict | None:
         full = self._json(f"/projects/{_pid(full_name)}/pipelines/{pipeline_id}")
         return full if isinstance(full, dict) else listing[0]
 
-    def pipeline_jobs(self, full_name: str, pipeline_id: int) -> list[dict]:
+    def pipeline_jobs(self, full_name: str, pipeline_id: int) -> list[dict[str, Any]]:
         """Every job in one pipeline.
 
         A GitLab pipeline is one run containing many jobs, where GitHub has many
@@ -350,7 +352,7 @@ def _checks_state(status: str | None) -> str:
     return "none"
 
 
-def _visibility(raw: dict) -> str:
+def _visibility(raw: dict[str, Any]) -> str:
     """The project's visibility: ``public``, ``internal`` or ``private``.
 
     Named rather than inlined because ``public_only`` tests it too, and
@@ -360,7 +362,7 @@ def _visibility(raw: dict) -> str:
     return raw.get("visibility") or "unknown"
 
 
-def _protection(entry: dict | None) -> tuple[bool | None, bool, int]:
+def _protection(entry: dict[str, Any] | None) -> tuple[bool | None, bool, int]:
     """``(protected, allows_force_push, required_reviews)`` from one entry.
 
     ``protected`` is never None on GitLab. GitHub's third state exists because
@@ -421,7 +423,7 @@ def collect(
     )
     projects = [r for r in listing if r["path_with_namespace"] not in excluded]
 
-    def one(raw: dict) -> RemoteRepo:
+    def one(raw: dict[str, Any]) -> RemoteRepo:
         full_name = raw["path_with_namespace"]
         branch = raw.get("default_branch") or "main"
         head_sha = api.branch_sha(full_name, branch)

collector/jq_collector/metrics.py

@@ -30,18 +30,20 @@
 
 from __future__ import annotations
 
-from prometheus_client.core import CounterMetricFamily, GaugeMetricFamily
+from collections.abc import Iterator
+
+from prometheus_client.core import CounterMetricFamily, GaugeMetricFamily, Metric
 
 from .forge import GOOD_CONCLUSIONS as _GOOD_CONCLUSIONS
 from .forge import INCONCLUSIVE_CONCLUSIONS as _INCONCLUSIVE_CONCLUSIONS
-from .state import LocalRepo, RemoteRepo, Snapshot, WorkflowRun
+from .state import LocalRepo, RemoteRepo, Snapshot, Store, WorkflowRun
 
 
 def _gauge(name: str, doc: str, labels: list[str] | None = None) -> GaugeMetricFamily:
     return GaugeMetricFamily(name, doc, labels=labels or [])
 
 
-def render(snap: Snapshot):
+def render(snap: Snapshot) -> Iterator[Metric]:
     yield from _health(snap)
 
     f = _Families()
@@ -58,7 +60,7 @@ def render(snap: Snapshot):
     yield from f.in_exposition_order()
 
 
-def _health(snap: Snapshot):
+def _health(snap: Snapshot) -> Iterator[Metric]:
     """The fleet-wide families: collector health, the rate limit, the newest template."""
     # -- collector health ------------------------------------------------
     last_success = _gauge(
@@ -582,8 +584,8 @@ def _add_size(f: _Families, ident: list[str], local: LocalRepo) -> None:
 class FleetCollector:
     """Adapts the store to prometheus_client's collector interface."""
 
-    def __init__(self, store) -> None:
+    def __init__(self, store: Store) -> None:
         self._store = store
 
-    def collect(self):
+    def collect(self) -> Iterator[Metric]:
         yield from render(self._store.snapshot())

collector/jq_collector/repos.py

@@ -75,7 +75,7 @@ def _is_checkout(path: str) -> bool:
     return os.path.exists(os.path.join(path, ".git"))
 
 
-def _declared_forge(item: dict, index: int) -> str | None:
+def _declared_forge(item: dict[str, Any], index: int) -> str | None:
     """The entry's explicit ``forge:``, validated, or None if it has none.
 
     An unknown value is refused rather than defaulted. Silently reading
@@ -92,7 +92,7 @@ def _declared_forge(item: dict, index: int) -> str | None:
     return declared
 
 
-def _excluded(item: dict, index: int, raw: str) -> frozenset[str]:
+def _excluded(item: dict[str, Any], index: int, raw: str) -> frozenset[str]:
     """A folder's ``exclude:`` list, as a set of directory names.
 
     A single name may be written on its own - ``exclude: numpy`` - because a
@@ -121,7 +121,9 @@ def _excluded(item: dict, index: int, raw: str) -> frozenset[str]:
     return frozenset(names - {""})
 
 
-def _folder(item: dict, index: int, host_root: str, claimed: frozenset[str]) -> list[dict]:
+def _folder(
+    item: dict[str, Any], index: int, host_root: str, claimed: frozenset[str]
+) -> list[dict[str, Any]]:
     """A ``folder:`` entry -> one ``path:`` entry per checkout inside it.
 
     Only the folder's own children are looked at, never their children in turn.
@@ -240,7 +242,7 @@ def _expand(
     return [(item, False)]
 
 
-def _claimed_paths(entries: list, host_root: str) -> frozenset[str]:
+def _claimed_paths(entries: list[Any], host_root: str) -> frozenset[str]:
     """Every checkout the file names by ``path``, resolved for this filesystem.
 
     A folder skips these, so an entry written out for one checkout inside a
@@ -311,7 +313,7 @@ def _entry(item: Any, index: int, host_root: str) -> tuple[str, str | None, str]
     return parsed.full_name, path, declared or parsed.forge
 
 
-def _read_entries(source: str) -> list:
+def _read_entries(source: str) -> list[Any]:
     """The non-empty ``repos:`` list of ``source``, or a FleetError saying why not."""
     import yaml
 

collector/pyproject.toml

@@ -40,11 +40,13 @@ source = ["jq_collector"]
 fail_under = 100
 
 [tool.mypy]
-# The package only, checked against the oldest Python it claims to run on. Not
-# --strict yet: this is the floor that catches an undefined attribute before a
-# refresh does, and it can be tightened once it holds.
+# The package only, checked against the oldest Python it claims to run on, and
+# strict: every signature typed, every generic parameterised. JSON payloads are
+# dict[str, Any] - the forge decides their shape, so Any is the honest type
+# there rather than a gap. CI's Types step reads this, so it enforces it too.
 files = ["jq_collector"]
 python_version = "3.11"
+strict = true
 
 [tool.ruff]
 # Kept here rather than passed as a flag, so `check` and `format` cannot