fix: enable strict mypy type checking for timeout-sampler

.pre-commit-config.yaml

@@ -60,4 +60,4 @@ repos:
     hooks:
       - id: mypy
         exclude: (tests/)
-        additional_dependencies: [types-requests, types-PyYAML]
+        additional_dependencies: [types-requests, types-PyYAML, types-colorama]

pyproject.toml

@@ -34,7 +34,13 @@ warn_redundant_casts = true
 packages = ["webhook_server"]
 
 [tool.uv]
-dev-dependencies = ["ipdb>=0.13.13", "ipython>=8.12.3"]
+dev-dependencies = [
+    "ipdb>=0.13.13",
+    "ipython>=8.12.3",
+    "types-colorama>=0.4.15.20240311",
+    "types-pyyaml>=6.0.12.20250516",
+    "types-requests>=2.32.4.20250611",
+]
 
 [project]
 name = "github-webhook-server"

webhook_server/libs/issue_comment_handler.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import asyncio
-from typing import TYPE_CHECKING, Callable
+from typing import TYPE_CHECKING, Callable, Any, Coroutine, Union
+from asyncio import Task
 
 from github.PullRequest import PullRequest
 
@@ -298,7 +299,7 @@ async def process_retest_command(self, pull_request: PullRequest, command_args:
             await asyncio.to_thread(pull_request.create_issue_comment, msg)
 
         if _supported_retests:
-            tasks = []
+            tasks: list[Union[Coroutine[Any, Any, Any], Task[Any]]] = []
             for _test in _supported_retests:
                 self.logger.debug(f"{self.log_prefix} running retest {_test}")
                 task = asyncio.create_task(_retests_to_func_map[_test](pull_request=pull_request))

webhook_server/libs/owners_files_handler.py

@@ -1,6 +1,6 @@
 import asyncio
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Coroutine
 
 import yaml
 from asyncstdlib import functools
@@ -90,7 +90,7 @@ async def _get_file_content(self, content_path: str, pull_request: PullRequest)
     async def get_all_repository_approvers_and_reviewers(self, pull_request: PullRequest) -> dict[str, dict[str, Any]]:
         # Dictionary mapping OWNERS file paths to their approvers and reviewers
         _owners: dict[str, dict[str, Any]] = {}
-        tasks = []
+        tasks: list[Coroutine[Any, Any, Any]] = []
 
         max_owners_files = 1000  # Configurable limit
         owners_count = 0

webhook_server/libs/pull_request_handler.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Coroutine
 
 from github.PullRequest import PullRequest
 
@@ -63,7 +63,7 @@ async def process_pull_request_webhook_data(self, pull_request: PullRequest) ->
             await self.set_wip_label_based_on_title(pull_request=pull_request)
 
         if hook_action in ("opened", "reopened", "ready_for_review"):
-            tasks = []
+            tasks: list[Coroutine[Any, Any, Any]] = []
 
             if hook_action in ("opened", "ready_for_review"):
                 welcome_msg = self._prepare_welcome_comment()
@@ -76,18 +76,18 @@ async def process_pull_request_webhook_data(self, pull_request: PullRequest) ->
             results = await asyncio.gather(*tasks, return_exceptions=True)
             for result in results:
                 if isinstance(result, Exception):
-                    self.logger.error(f"{self.log_prefix} Async task failed: {results}")
+                    self.logger.error(f"{self.log_prefix} Async task failed: {result}")
 
             # Set auto merge only after all initialization of a new PR is done.
             await self.set_pull_request_automerge(pull_request=pull_request)
 
         if hook_action == "synchronize":
-            tasks = []
+            sync_tasks: list[Coroutine[Any, Any, Any]] = []
 
-            tasks.append(self.process_opened_or_synchronize_pull_request(pull_request=pull_request))
-            tasks.append(self.remove_labels_when_pull_request_sync(pull_request=pull_request))
+            sync_tasks.append(self.process_opened_or_synchronize_pull_request(pull_request=pull_request))
+            sync_tasks.append(self.remove_labels_when_pull_request_sync(pull_request=pull_request))
 
-            results = await asyncio.gather(*tasks, return_exceptions=True)
+            results = await asyncio.gather(*sync_tasks, return_exceptions=True)
 
             for result in results:
                 if isinstance(result, Exception):
@@ -130,9 +130,9 @@ async def process_pull_request_webhook_data(self, pull_request: PullRequest) ->
             _split_label = labeled.split(LABELS_SEPARATOR, 1)
 
             if len(_split_label) == 2:
-                _lable_prefix, _user = _split_label
+                _label_prefix, _user = _split_label
 
-                if f"{_lable_prefix}{LABELS_SEPARATOR}" in (
+                if f"{_label_prefix}{LABELS_SEPARATOR}" in (
                     APPROVED_BY_LABEL_PREFIX,
                     LGTM_BY_LABEL_PREFIX,
                     CHANGED_REQUESTED_BY_LABEL_PREFIX,
@@ -395,15 +395,13 @@ async def close_issue_for_merged_or_closed_pr(self, pull_request: PullRequest, h
                 break
 
     async def process_opened_or_synchronize_pull_request(self, pull_request: PullRequest) -> None:
-        tasks = []
+        tasks: list[Coroutine[Any, Any, Any]] = []
 
         tasks.append(self.owners_file_handler.assign_reviewers(pull_request=pull_request))
         tasks.append(
             self.labels_handler._add_label(
-                **{
-                    "pull_request": pull_request,
-                    "label": f"{BRANCH_LABEL_PREFIX}{pull_request.base.ref}",
-                },
+                pull_request=pull_request,
+                label=f"{BRANCH_LABEL_PREFIX}{pull_request.base.ref}",
             )
         )
         tasks.append(self.label_pull_request_by_merge_state(pull_request=pull_request))
@@ -477,7 +475,7 @@ async def set_pull_request_automerge(self, pull_request: PullRequest) -> None:
                 self.logger.error(f"{self.log_prefix} Exception while setting auto merge: {exp}")
 
     async def remove_labels_when_pull_request_sync(self, pull_request: PullRequest) -> None:
-        tasks = []
+        tasks: list[Coroutine[Any, Any, Any]] = []
         for _label in pull_request.labels:
             _label_name = _label.name
             if (
@@ -488,10 +486,8 @@ async def remove_labels_when_pull_request_sync(self, pull_request: PullRequest)
             ):
                 tasks.append(
                     self.labels_handler._remove_label(
-                        **{
-                            "pull_request": pull_request,
-                            "label": _label_name,
-                        },
+                        pull_request=pull_request,
+                        label=_label_name,
                     )
                 )
 
@@ -601,7 +597,7 @@ async def check_if_can_be_merged(self, pull_request: PullRequest) -> None:
             if required_check_failed_failure_output:
                 failure_output += required_check_failed_failure_output
 
-            labels_failure_output = self._check_lables_for_can_be_merged(labels=_labels)
+            labels_failure_output = self._check_labels_for_can_be_merged(labels=_labels)
             if labels_failure_output:
                 failure_output += labels_failure_output
 
@@ -699,8 +695,8 @@ async def _check_if_pr_approved(self, labels: list[str]) -> str:
 
         return error
 
-    def _check_lables_for_can_be_merged(self, labels: list[str]) -> str:
-        self.logger.debug(f"{self.log_prefix} _check_lables_for_can_be_merged.")
+    def _check_labels_for_can_be_merged(self, labels: list[str]) -> str:
+        self.logger.debug(f"{self.log_prefix} _check_labels_for_can_be_merged.")
         failure_output = ""
 
         for _label in labels: