ref(typing)!: enforce strict mypy across Sentry's Python codebase

.github/workflows/pre-commit.yml

@@ -49,15 +49,7 @@ jobs:
         uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
         with:
           token: ${{ github.token }}
-          # Enable listing of files matching each filter.
-          # Paths to files will be available in `${FILTER_NAME}_files` output variable.
-          list-files: json
-
-          # It doesn't make sense to lint deleted files.
-          # Therefore we specify we are only interested in added or modified files.
           filters: |
-            all:
-              - added|modified: '**/*'
             frontend:
               - 'static/**'
               - 'tests/js/**'
@@ -94,10 +86,18 @@ jobs:
       - name: Run prek on PR commits
         id: prek
         env:
-          ALL_FILES: ${{ steps.changes.outputs.all_files }}
+          GH_TOKEN: ${{ github.token }}
+          REPOSITORY: ${{ github.repository }}
+          PULL_REQUEST_NUMBER: ${{ github.event.pull_request.number }}
+          HEAD_SHA: ${{ github.sha }}
         run: |
-          jq '.[]' --raw-output <<< "$ALL_FILES" |
-            xargs .venv/bin/prek run -q --files
+          if [[ -n "$PULL_REQUEST_NUMBER" ]]; then
+            gh api --paginate "repos/$REPOSITORY/pulls/$PULL_REQUEST_NUMBER/files?per_page=100" \
+              --jq '.[] | select(.status != "removed") | .filename'
+          else
+            git fetch --no-tags --depth=2 origin "$HEAD_SHA"
+            git diff --name-only --diff-filter=AM "$HEAD_SHA^" "$HEAD_SHA"
+          fi | xargs .venv/bin/prek run -q --files
 
       - name: Prune suppression budgets
         if: github.event_name == 'pull_request' && !cancelled() && (steps.prek.outcome == 'success' || steps.prek.outcome == 'failure') && steps.changes.outputs.frontend == 'true'

.pre-commit-config.yaml

@@ -75,6 +75,12 @@ repos:
         entry: python3 -m tools.mypy_helpers.prevent_weaklist_additions
         files: ^pyproject\.toml$
         language: python
+      - id: check-typing-campaign
+        name: check typing campaign gates
+        entry: python3 -m tools.mypy_helpers.check_typing_campaign
+        language: python
+        types: [python]
+        require_serial: true
       - id: prevent-push
         name: prevent pushing master
         stages: [pre-push]

fixtures/apidocs_test_case.py

@@ -3,10 +3,13 @@
 
 import orjson
 from django.conf import settings
+from django.http import HttpRequest
 from openapi_core.contrib.django import DjangoOpenAPIRequest, DjangoOpenAPIResponse
 from openapi_core.spec import Spec
 from openapi_core.validation.response.validators import V30ResponseDataValidator
+from rest_framework.response import Response
 
+from sentry.services.eventstore.models import Event
 from sentry.testutils.cases import APITestCase
 from sentry.testutils.helpers.datetime import before_now
 from sentry.testutils.skips import requires_snuba
@@ -30,7 +33,7 @@ class APIDocsTestCase(APITestCase):
     def cached_schema(self) -> Spec:
         return _load_schema()
 
-    def validate_schema(self, request, response):
+    def validate_schema(self, request: HttpRequest, response: Response) -> None:
         assert 200 <= response.status_code < 300, response.status_code
 
         if isinstance(response.data, list):
@@ -42,9 +45,9 @@ def validate_schema(self, request, response):
             DjangoOpenAPIResponse(response),  # type: ignore[arg-type]  # Werkzeug 3.1 Headers vs Mapping
         )
 
-    def create_event(self, name, **kwargs):
+    def create_event(self, name: str, **kwargs: object) -> Event:
         # Somewhat sane default data.
-        data = {
+        data: dict[str, object] = {
             "event_id": (name * 32)[:32],
             "fingerprint": ["1"],
             "sdk": {"version": "5.17.0", "name": "sentry.javascript.browser"},

fixtures/gitlab.py

@@ -18,7 +18,7 @@
 class GitLabTestCase(APITestCase):
     provider = IntegrationProviderSlug.GITLAB.value
 
-    def setUp(self):
+    def setUp(self) -> None:
         self.login_as(self.user)
         with assume_test_silo_mode(SiloMode.CONTROL):
             self.integration = Integration.objects.create(

fixtures/integrations/jira/mock.py

@@ -1,11 +1,14 @@
+from collections.abc import Mapping
+from typing import Any, cast
+
 from fixtures.integrations.jira.stub_client import StubJiraApiClient
 from fixtures.integrations.mock_service import MockService
 
 DEFAULT_PROJECT_ID = "10000"
 
 
 class MockJira(StubJiraApiClient, MockService):
-    def get_projects_list(self, cached: bool = True):
+    def get_projects_list(self, cached: bool = True) -> list[dict[str, Any]]:
         """
         List all projects in the Jira data format.
 
@@ -25,10 +28,10 @@ def get_projects_list(self, cached: bool = True):
                 },
                 "simplified": False,
             }
-            for project_name in self._get_project_names() + [DEFAULT_PROJECT_ID]
+            for project_name in [*self._get_project_names(), DEFAULT_PROJECT_ID]
         ]
 
-    def set_createmeta(self, project, createmeta):
+    def set_createmeta(self, project: str, createmeta: object) -> None:
         """
         This special method lets you seed the stub data with your own metadata.
         # TODO validate createmeta
@@ -39,7 +42,7 @@ def set_createmeta(self, project, createmeta):
         """
         return self._set_data(project, "createmeta", createmeta)
 
-    def get_create_meta_for_project(self, project):
+    def get_create_meta_for_project(self, project: str) -> dict[str, Any] | None:
         """
         Get the Jira "createmeta" for a project.
 
@@ -50,12 +53,13 @@ def get_create_meta_for_project(self, project):
 
         createmeta = self._get_data(project, "createmeta")
         if createmeta:
-            return createmeta
+            # justification: seeded fixture data is JSON written through `set_createmeta`.
+            return cast("dict[str, Any]", createmeta)
 
         # Fallback to stub data
         return super().get_create_meta_for_project(project)
 
-    def create_issue(self, raw_form_data):
+    def create_issue(self, raw_form_data: Mapping[str, object]) -> dict[str, str]:
         """
         Create a new Jira issue. Currently overwrites if the issue already exists.
 
@@ -64,13 +68,15 @@ def create_issue(self, raw_form_data):
         """
         self._throw_if_broken()
 
-        project = raw_form_data.get("project", {}).get("id")
-        ticket_key = self._get_new_ticket_name(project)
-        self._set_data(project, ticket_key, {"fields": raw_form_data})
+        project = cast("Mapping[str, object]", raw_form_data.get("project", {})).get("id")
+        # justification: Jira project ids are strings in the request fixture payload.
+        project_id = cast("str", project)
+        ticket_key = self._get_new_ticket_name(project_id)
+        self._set_data(project_id, ticket_key, {"fields": raw_form_data})
 
         return {"key": ticket_key}
 
-    def get_issue(self, issue_key):
+    def get_issue(self, issue_key: str) -> dict[str, Any] | None:
         """
         Get a saved issue from Jira.
 
@@ -82,10 +88,12 @@ def get_issue(self, issue_key):
         if not data:
             return None
 
-        data.update(
+        # justification: seeded fixture data is JSON written through `_set_data`.
+        issue = cast("dict[str, Any]", data)
+        issue.update(
             {
                 "id": issue_key,
                 "key": issue_key,
             }
         )
-        return data
+        return issue

fixtures/integrations/jira/stub_client.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from typing import Any
 
 from fixtures.integrations.stub_service import StubService
@@ -7,59 +8,71 @@
 class StubJiraApiClient(StubService):
     service_name = "jira"
 
-    def get_create_meta_for_project(self, project):
-        response = self._get_stub_data("createmeta_response.json")
+    def get_create_meta_for_project(self, project: str) -> dict[str, Any] | None:
+        response: dict[str, Any] = self._get_stub_data("createmeta_response.json")
         if project == "10001":
             response["projects"][0]["id"] = "10001"
-        return response["projects"][0]
+        project_data: dict[str, Any] | None = response["projects"][0]
+        return project_data
 
-    def get_issue_fields(self, project_id, issue_type_id):
-        return self._get_stub_data("issue_fields_response.json")
+    def get_issue_fields(self, project_id: str, issue_type_id: str) -> dict[str, Any]:
+        response: dict[str, Any] = self._get_stub_data("issue_fields_response.json")
+        return response
 
-    def get_issue_types(self, project_id):
-        return self._get_stub_data("issue_types_response.json")
+    def get_issue_types(self, project_id: str) -> dict[str, Any]:
+        response: dict[str, Any] = self._get_stub_data("issue_types_response.json")
+        return response
 
-    def get_priorities(self):
-        return self._get_stub_data("priorities_response.json")
+    def get_priorities(self) -> list[dict[str, Any]]:
+        response: list[dict[str, Any]] = self._get_stub_data("priorities_response.json")
+        return response
 
-    def get_versions(self, project_id):
-        return self._get_stub_data("versions_response.json")
+    def get_versions(self, project_id: str) -> list[dict[str, Any]]:
+        response: list[dict[str, Any]] = self._get_stub_data("versions_response.json")
+        return response
 
-    def get_projects_paginated(self, params: dict[str, Any] | None = None):
-        return self._get_stub_data("projects_paginated.json")
+    def get_projects_paginated(self, params: dict[str, Any] | None = None) -> dict[str, Any]:
+        response: dict[str, Any] = self._get_stub_data("projects_paginated.json")
+        return response
 
-    def get_projects_list(self, cached: bool = True):
-        return self._get_stub_data("project_list_response.json")
+    def get_projects_list(self, cached: bool = True) -> list[dict[str, Any]]:
+        response: list[dict[str, Any]] = self._get_stub_data("project_list_response.json")
+        return response
 
-    def get_issue(self, issue_key):
-        return self._get_stub_data("get_issue_response.json")
+    def get_issue(self, issue_key: str) -> dict[str, Any] | None:
+        response: dict[str, Any] | None = self._get_stub_data("get_issue_response.json")
+        return response
 
-    def create_comment(self, issue_id, comment):
+    def create_comment(self, issue_id: str, comment: str) -> str:
         return comment
 
-    def update_comment(self, issue_key, comment_id, comment):
+    def update_comment(self, issue_key: str, comment_id: str, comment: str) -> str:
         return comment
 
-    def create_issue(self, raw_form_data):
+    def create_issue(self, raw_form_data: Mapping[str, Any]) -> dict[str, str]:
         return {"key": "APP-123"}
 
-    def get_transitions(self, issue_key):
-        return self._get_stub_data("transition_response.json")["transitions"]
+    def get_transitions(self, issue_key: str) -> list[dict[str, Any]]:
+        response: dict[str, Any] = self._get_stub_data("transition_response.json")
+        transitions: list[dict[str, Any]] = response["transitions"]
+        return transitions
 
-    def transition_issue(self, issue_key, transition_id):
+    def transition_issue(self, issue_key: str, transition_id: str) -> None:
         pass
 
     def user_id_field(self) -> str:
         return "accountId"
 
-    def get_user(self, user_id):
-        user = self._get_stub_data("user.json")
+    def get_user(self, user_id: str) -> dict[str, Any]:
+        user: dict[str, Any] = self._get_stub_data("user.json")
         if user["accountId"] == user_id:
             return user
         raise ApiError("no user found")
 
-    def get_valid_statuses(self):
-        return self._get_stub_data("status_response.json")
+    def get_valid_statuses(self) -> list[dict[str, Any]]:
+        response: list[dict[str, Any]] = self._get_stub_data("status_response.json")
+        return response
 
-    def search_users_for_project(self, project, username):
-        return [self._get_stub_data("user.json")]
+    def search_users_for_project(self, project: str, username: str) -> list[dict[str, Any]]:
+        user: dict[str, Any] = self._get_stub_data("user.json")
+        return [user]

fixtures/integrations/mock_service.py

@@ -3,6 +3,7 @@
 import os
 import shutil
 from collections import defaultdict
+from collections.abc import KeysView
 from typing import Any
 
 import orjson
@@ -20,24 +21,25 @@ class MockService(StubService):
     Like stubs, mocks can make tests simpler and more reliable.
     """
 
-    def __init__(self, mode="memory"):
+    def __init__(self, mode: str = "memory") -> None:
         """
         Initialize the mock instance. Wipe the previous instance's data if it exists.
         """
         super().__init__()
         self.mode = mode
-        self._next_error_code = None
-        self._next_ids = defaultdict(int)
+        self._next_error_code: int | None = None
+        self._next_ids: defaultdict[str, int] = defaultdict(int)
+        self._memory: defaultdict[str, dict[str, Any]]
 
         if self.mode == "file":
             path = os.path.join(FIXTURE_DIRECTORY, self.service_name, "data")
             if os.path.exists(path):
                 shutil.rmtree(path)
             os.makedirs(path)
         else:
-            self._memory: dict[str, dict[str, Any]] = defaultdict(dict)
+            self._memory = defaultdict(dict)
 
-    def add_project(self, project):
+    def add_project(self, project: str) -> None:
         """
         Create a new, empty project.
 
@@ -48,7 +50,7 @@ def add_project(self, project):
         if self.mode == "file":
             self._get_project_path(project)
 
-    def remove_project(self, project):
+    def remove_project(self, project: str) -> None:
         """
         Totally wipe out a project.
 
@@ -60,13 +62,13 @@ def remove_project(self, project):
             path = self._get_project_path(project)
             shutil.rmtree(path)
 
-    def break_next_api_call(self, error_code=500):
+    def break_next_api_call(self, error_code: int = 500) -> None:
         """
         Simulate an outage for a single API call.
         """
         self._next_error_code = error_code
 
-    def _throw_if_broken(self, message_option=None):
+    def _throw_if_broken(self, message_option: str | None = None) -> None:
         """
         See break_next_api_call.
         :param message_option: What should the message be if this raises?
@@ -77,22 +79,22 @@ def _throw_if_broken(self, message_option=None):
             message = message_option or f"{self.service_name} is down"
             raise Exception(f"{self._next_error_code}: {message}")
 
-    def _get_project_names(self):
+    def _get_project_names(self) -> KeysView[str]:
         return self._next_ids.keys()
 
-    def _get_new_ticket_name(self, project):
+    def _get_new_ticket_name(self, project: str) -> str:
         counter = self._next_ids[project]
         self._next_ids[project] = counter + 1
 
         return f"{project}-{base32_encode(counter)}"
 
-    def _get_project_path(self, project):
+    def _get_project_path(self, project: str) -> str:
         path = os.path.join(FIXTURE_DIRECTORY, self.service_name, "data", project)
 
         os.makedirs(path, exist_ok=True)
         return path
 
-    def _set_data(self, project, name, data):
+    def _set_data(self, project: str, name: str, data: object) -> None:
         if self.mode == "memory":
             if not self._memory[project]:
                 self._memory[project] = defaultdict()
@@ -103,7 +105,7 @@ def _set_data(self, project, name, data):
         with open(path, "wb") as f:
             f.write(orjson.dumps(data))
 
-    def _get_data(self, project, name):
+    def _get_data(self, project: str, name: str) -> object | None:
         if self.mode == "memory":
             if not self._memory[project]:
                 self._memory[project] = defaultdict()
@@ -114,4 +116,5 @@ def _get_data(self, project, name):
             return None
 
         with open(path, "rb") as f:
-            return orjson.loads(f.read())
+            data: object = orjson.loads(f.read())
+            return data