Fix typing

Author: ewjoachim

coverage_comment/__main__.py

@@ -3,7 +3,7 @@
 from coverage_comment import main
 
 
-def main_call(name):
+def main_call(name: str):
     if name == "__main__":
         main.main()
 

coverage_comment/activity.py

@@ -16,7 +16,7 @@ class ActivityNotFound(Exception):
 def find_activity(
     event_name: str,
     is_default_branch: bool,
-    event_type: str,
+    event_type: str | None,
     is_pr_merged: bool,
 ) -> str:
     """Find the activity to perform based on the event type and payload."""

coverage_comment/coverage.py

@@ -6,6 +6,7 @@
 import json
 import pathlib
 from collections.abc import Sequence
+from typing import Any
 
 from coverage_comment import log, subprocess
 
@@ -21,20 +22,19 @@ class CoverageMetadata:
     show_contexts: bool
 
 
-class OutputMixin:
-    def as_output(self, prefix: str) -> dict:
-        data = dataclasses.asdict(self)
-        output = {}
-        for key, value in data.items():
-            if value is not None and not isinstance(value, dict):
-                output[f"{prefix}_{key}"] = (
-                    float(value) if isinstance(value, decimal.Decimal) else value
-                )
-        return output
+def as_output(obj: Any, prefix: str) -> dict[str, Any]:
+    data = dataclasses.asdict(obj)
+    output: dict[str, Any] = {}
+    for key, value in data.items():
+        if value is not None and not isinstance(value, dict):
+            output[f"{prefix}_{key}"] = (
+                float(value) if isinstance(value, decimal.Decimal) else value
+            )
+    return output
 
 
 @dataclasses.dataclass(kw_only=True)
-class CoverageInfo(OutputMixin):
+class CoverageInfo:
     covered_lines: int
     num_statements: int
     percent_covered: decimal.Decimal
@@ -88,7 +88,7 @@ def violation_lines(self) -> list[int]:
 
 
 @dataclasses.dataclass(kw_only=True)
-class DiffCoverage(OutputMixin):
+class DiffCoverage:
     total_num_lines: int
     total_num_violations: int
     total_percent_covered: decimal.Decimal
@@ -112,7 +112,7 @@ def compute_coverage(
 
 def get_coverage_info(
     merge: bool, coverage_path: pathlib.Path
-) -> tuple[dict, Coverage]:
+) -> tuple[dict[str, Any], Coverage]:
     try:
         if merge:
             subprocess.run("coverage", "combine", path=coverage_path)
@@ -160,7 +160,7 @@ def generate_coverage_markdown(coverage_path: pathlib.Path) -> str:
     )
 
 
-def _make_coverage_info(data: dict) -> CoverageInfo:
+def _make_coverage_info(data: dict[str, Any]) -> CoverageInfo:
     """Build a CoverageInfo object from a "summary" or "totals" key."""
     return CoverageInfo(
         covered_lines=data["covered_lines"],
@@ -180,7 +180,7 @@ def _make_coverage_info(data: dict) -> CoverageInfo:
     )
 
 
-def extract_info(data: dict, coverage_path: pathlib.Path) -> Coverage:
+def extract_info(data: dict[str, Any], coverage_path: pathlib.Path) -> Coverage:
     """
     {
         "meta": {
@@ -246,7 +246,7 @@ def extract_info(data: dict, coverage_path: pathlib.Path) -> Coverage:
 def get_diff_coverage_info(
     added_lines: dict[pathlib.Path, list[int]], coverage: Coverage
 ) -> DiffCoverage:
-    files = {}
+    files: dict[pathlib.Path, FileDiffCoverage] = {}
     total_num_lines = 0
     total_num_violations = 0
     num_changed_lines = 0

coverage_comment/files.py

@@ -11,8 +11,7 @@
 import pathlib
 import shutil
 import tempfile
-from collections.abc import Callable
-from typing import Protocol, TypedDict
+from typing import Any, Protocol, TypedDict
 
 import httpx
 
@@ -60,7 +59,7 @@ def apply(self):
 
 def compute_files(
     line_rate: decimal.Decimal,
-    raw_coverage_data: dict,
+    raw_coverage_data: dict[str, Any],
     coverage_path: pathlib.Path,
     minimum_green: decimal.Decimal,
     minimum_orange: decimal.Decimal,
@@ -97,7 +96,9 @@ def compute_files(
 
 
 def compute_datafile(
-    raw_coverage_data: dict, line_rate: decimal.Decimal, coverage_path: pathlib.Path
+    raw_coverage_data: dict[str, Any],
+    line_rate: decimal.Decimal,
+    coverage_path: pathlib.Path,
 ) -> str:
     return json.dumps(
         {
@@ -108,7 +109,7 @@ def compute_datafile(
     )
 
 
-def parse_datafile(contents) -> tuple[coverage.Coverage | None, decimal.Decimal]:
+def parse_datafile(contents: str) -> tuple[coverage.Coverage | None, decimal.Decimal]:
     file_contents = json.loads(contents)
     coverage_rate = decimal.Decimal(str(file_contents["coverage"])) / decimal.Decimal(
         "100"
@@ -128,7 +129,11 @@ class ImageURLs(TypedDict):
     dynamic: str
 
 
-def get_urls(url_getter: Callable) -> ImageURLs:
+class URLGetter(Protocol):
+    def __call__(self, path: pathlib.Path) -> str: ...
+
+
+def get_urls(url_getter: URLGetter) -> ImageURLs:
     return {
         "direct": url_getter(path=BADGE_PATH),
         "endpoint": badge.get_endpoint_url(endpoint_url=url_getter(path=ENDPOINT_PATH)),
@@ -137,7 +142,9 @@ def get_urls(url_getter: Callable) -> ImageURLs:
 
 
 def get_coverage_html_files(
-    *, coverage_path: pathlib.Path, gen_dir: pathlib.Path = pathlib.Path("/tmp")
+    *,
+    coverage_path: pathlib.Path,
+    gen_dir: pathlib.Path | None = None,
 ) -> ReplaceDir:
     html_dir = pathlib.Path(tempfile.mkdtemp(dir=gen_dir))
     coverage.generate_coverage_html_files(

coverage_comment/github.py

@@ -7,6 +7,7 @@
 import re
 import sys
 import zipfile
+from collections.abc import Iterable
 from typing import Any
 from urllib.parse import urlparse
 
@@ -106,7 +107,9 @@ def download_artifact(
         raise NoArtifact(f"File named {filename} not found in artifact {artifact_name}")
 
 
-def _fetch_artifacts(repo_path, run_id):
+def _fetch_artifacts(
+    repo_path: github_client.Endpoint, run_id: int
+) -> Iterable[github_client.JsonObject]:
     page = 1
     total_fetched = 0
 
@@ -143,31 +146,13 @@ def find_pr_for_branch(
 
     full_branch = f"{owner}:{branch}"
 
-    common_kwargs = {"head": full_branch, "sort": "updated", "direction": "desc"}
-    try:
-        return next(
-            iter(
-                pr.number
-                for pr in github.repos(repository).pulls.get(
-                    state="open", **common_kwargs
-                )
-            )
-        )
-    except StopIteration:
-        pass
-    log.info(f"No open PR found for branch {branch}, defaulting to all PRs")
+    for state in ["open", "all"]:
+        for pr in github.repos(repository).pulls.get(
+            state=state, head=full_branch, sort="updated", direction="desc"
+        ):
+            return pr.number  # pyright: ignore
 
-    try:
-        return next(
-            iter(
-                pr.number
-                for pr in github.repos(repository).pulls.get(
-                    state="all", **common_kwargs
-                )
-            )
-        )
-    except StopIteration:
-        raise CannotDeterminePR(f"No open PR found for branch {branch}")
+    raise CannotDeterminePR(f"No open PR found for branch {branch}")
 
 
 def get_my_login(github: github_client.GitHub) -> str:
@@ -195,10 +180,14 @@ def post_comment(
     comments_path = github.repos(repository).issues.comments
 
     for comment in issue_comments_path.get():
-        if comment.user.login == me and marker in comment.body:
+        login: str = comment.user.login  # pyright: ignore
+        body: str = comment.body  # pyright: ignore
+        comment_id: int = comment.id  # pyright: ignore
+
+        if login == me and marker in body:
             log.info("Update previous comment")
             try:
-                comments_path(comment.id).patch(body=contents)
+                comments_path(comment_id).patch(body=contents)
             except github_client.Forbidden as exc:
                 raise CannotPostComment from exc
             break
@@ -296,7 +285,7 @@ def get_pr_diff(github: github_client.GitHub, repository: str, pr_number: int) -
     return (
         github.repos(repository)
         .pulls(pr_number)
-        .get(headers={"Accept": "application/vnd.github.v3.diff"})
+        .get(headers={"Accept": "application/vnd.github.v3.diff"}, text=True)
     )
 
 
@@ -309,5 +298,5 @@ def get_branch_diff(
     return (
         github.repos(repository)
         .compare(f"{base_branch}...{head_branch}")
-        .get(headers={"Accept": "application/vnd.github.v3.diff"})
+        .get(headers={"Accept": "application/vnd.github.v3.diff"}, text=True)
     )