From b5ff77ea897d9adb61dd28a7a0f457403e7492af Mon Sep 17 00:00:00 2001 From: Thomas Schmelzer Date: Fri, 25 Sep 2026 16:46:51 +0400 Subject: [PATCH] fix: type-check jq_collector under mypy strict (#52) --- collector/jq_collector/github.py | 27 +++++++++++++++------------ collector/jq_collector/gitlab.py | 24 +++++++++++++----------- collector/jq_collector/metrics.py | 14 ++++++++------ collector/jq_collector/repos.py | 12 +++++++----- collector/pyproject.toml | 8 +++++--- 5 files changed, 48 insertions(+), 37 deletions(-) diff --git a/collector/jq_collector/github.py b/collector/jq_collector/github.py index f01f687..b276431 100644 --- a/collector/jq_collector/github.py +++ b/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" diff --git a/collector/jq_collector/gitlab.py b/collector/jq_collector/gitlab.py index d6fe085..693aac9 100644 --- a/collector/jq_collector/gitlab.py +++ b/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) diff --git a/collector/jq_collector/metrics.py b/collector/jq_collector/metrics.py index c28b1ad..2c1730d 100644 --- a/collector/jq_collector/metrics.py +++ b/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()) diff --git a/collector/jq_collector/repos.py b/collector/jq_collector/repos.py index bf2f60b..8583f3e 100644 --- a/collector/jq_collector/repos.py +++ b/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 diff --git a/collector/pyproject.toml b/collector/pyproject.toml index 30e9098..a88a165 100644 --- a/collector/pyproject.toml +++ b/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