Fix typing warnings

- and enable failOnWarnings for basedpyright

Author: stephanme

orgs/org_management/org_management.py

@@ -10,7 +10,7 @@
 import os
 import re
 from pathlib import Path
-from typing import Any, final, override
+from typing import Any, TextIO, final, override
 
 import jsonschema
 import yaml
@@ -22,10 +22,10 @@
 # https://yaml.org/spec/1.2.2/ requires unique keys
 class UniqueKeyLoader(yaml.SafeLoader):
     @override
-    def construct_mapping(self, node, deep=False):
-        mapping = set()
+    def construct_mapping(self, node: Any, deep: bool = False):
+        mapping = set[str]()
         for key_node, _ in node.value:
-            key = self.construct_object(key_node, deep=deep)
+            key: str = str(self.construct_object(key_node, deep=deep))  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]
             if key in mapping:
                 raise yaml.MarkedYAMLError(f"Duplicate key {key!r} found.", key_node.start_mark)
             mapping.add(key)
@@ -47,10 +47,12 @@ def __init__(
         working_groups: list[str] | None = None,
         branch_protection: str | None = None,
     ):
-        self.org_cfg = OrgGenerator._validate_github_org_cfg(OrgGenerator._yaml_load(static_org_cfg)) if static_org_cfg else {"orgs": {}}
+        self.org_cfg: dict[str, Any] = (
+            OrgGenerator._validate_github_org_cfg(OrgGenerator._yaml_load(static_org_cfg)) if static_org_cfg else {"orgs": {}}
+        )
         self.contributors = dict[str, set[str]]()
-        self.working_groups = {}
-        self.branch_protection = (
+        self.working_groups: dict[str, Any] = {}
+        self.branch_protection: dict[str, Any] = (
             OrgGenerator._validate_branch_protection(OrgGenerator._yaml_load(branch_protection))
             if branch_protection
             else {"branch-protection": {"orgs": {}}}
@@ -218,7 +220,7 @@ def write_branch_protection(self, path: str):
             return yaml.safe_dump(self.branch_protection, stream)
 
     @staticmethod
-    def _yaml_load(stream) -> dict[str, Any]:
+    def _yaml_load(stream: TextIO | str) -> dict[str, Any]:
         # safe_load + reject unique keys
         return yaml.load(stream, UniqueKeyLoader)
 
@@ -247,7 +249,7 @@ def _extract_wg_config(wg_charter: str):
         return OrgGenerator._validate_wg(OrgGenerator._yaml_load(match.group(1))) if match else None
 
     @staticmethod
-    def _empty_wg_config(name: str):
+    def _empty_wg_config(name: str) -> dict[str, Any]:
         return {
             "name": name,
             "org": OrgGenerator._DEFAULT_ORG,
@@ -258,7 +260,7 @@ def _empty_wg_config(name: str):
         }
 
     @staticmethod
-    def _wg_github_users(wg) -> set[str]:
+    def _wg_github_users(wg: dict[str, Any]) -> set[str]:
         users = {u["github"] for u in wg["execution_leads"]}
         users |= {u["github"] for u in wg["technical_leads"]}
         users |= {u["github"] for u in wg["bots"]}
@@ -271,7 +273,7 @@ def _wg_github_users(wg) -> set[str]:
         return users
 
     @staticmethod
-    def _wg_github_users_leads(wg) -> set[str]:
+    def _wg_github_users_leads(wg: dict[str, Any]) -> set[str]:
         users = {u["github"] for u in wg["execution_leads"]}
         users |= {u["github"] for u in wg["technical_leads"]}
         return users
@@ -297,7 +299,7 @@ def _wg_github_users_leads(wg) -> set[str]:
     }
 
     @staticmethod
-    def _validate_contributors(contributors) -> dict[str, Any]:
+    def _validate_contributors(contributors: dict[str, Any]) -> dict[str, Any]:
         jsonschema.validate(contributors, OrgGenerator._CONTRIBUTORS_SCHEMA)
         # check that orgs are in _ORGS
         for org in contributors["orgs"]:
@@ -343,7 +345,7 @@ def _validate_contributors(contributors) -> dict[str, Any]:
     }
 
     @staticmethod
-    def _validate_wg(wg) -> dict[str, Any]:
+    def _validate_wg(wg: dict[str, Any]) -> dict[str, Any]:
         jsonschema.validate(wg, OrgGenerator._WG_SCHEMA)
         # validate org and use 'cloudfoundry' if missing
         if "org" not in wg:
@@ -375,7 +377,7 @@ def _validate_wg(wg) -> dict[str, Any]:
     }
 
     @staticmethod
-    def _validate_github_org_cfg(cfg):
+    def _validate_github_org_cfg(cfg: dict[str, Any]) -> dict[str, Any]:
         jsonschema.validate(cfg, OrgGenerator._GITHUB_ORG_CFG_SCHEMA)
         # check that orgs are in _ORGS
         for org in cfg["orgs"]:
@@ -410,7 +412,7 @@ def _validate_github_org_cfg(cfg):
     }
 
     @staticmethod
-    def _validate_branch_protection(cfg):
+    def _validate_branch_protection(cfg: dict[str, Any]) -> dict[str, Any]:
         jsonschema.validate(cfg, OrgGenerator._BRANCH_PROTECTION_SCHEMA)
         # check that orgs are in _ORGS
         for org in cfg["branch-protection"]["orgs"]:
@@ -420,7 +422,7 @@ def _validate_branch_protection(cfg):
 
     # https://github.com/cloudfoundry/community/blob/main/toc/rfc/rfc-0005-github-teams-and-access.md
     @staticmethod
-    def _generate_wg_teams(wg) -> tuple[str, dict[str, Any]]:
+    def _generate_wg_teams(wg: dict[str, Any]) -> tuple[str, dict[str, Any]]:
         org = wg["org"]
         org_prefix = org + "/"
         org_prefix_len = len(org_prefix)
@@ -489,7 +491,7 @@ def _generate_wg_teams(wg) -> tuple[str, dict[str, Any]]:
         return (name, team)
 
     @staticmethod
-    def _generate_toc_team(wg) -> tuple[str, dict[str, Any]]:
+    def _generate_toc_team(wg: dict[str, Any]) -> tuple[str, dict[str, Any]]:
         org = wg["org"]
         org_prefix = org + "/"
         org_prefix_len = len(org_prefix)
@@ -504,7 +506,7 @@ def _generate_toc_team(wg) -> tuple[str, dict[str, Any]]:
         return ("toc", team)
 
     @staticmethod
-    def _generate_wg_leads_team(wgs: list[Any]) -> tuple[str, dict[str, Any]]:
+    def _generate_wg_leads_team(wgs: list[dict[str, Any]]) -> tuple[str, dict[str, Any]]:
         members = {u for wg in wgs for u in OrgGenerator._wg_github_users_leads(wg)}
         team = {
             "description": "Technical and Execution Leads for all WGs",
@@ -515,7 +517,7 @@ def _generate_wg_leads_team(wgs: list[Any]) -> tuple[str, dict[str, Any]]:
 
     # https://github.com/cloudfoundry/community/blob/main/toc/rfc/rfc-0015-branch-protection.md
     # returns hash with branch protection rules per repo
-    def _generate_wg_branch_protection(self, wg) -> dict[str, Any]:
+    def _generate_wg_branch_protection(self, wg: dict[str, Any]) -> dict[str, Any]:
         org = wg["org"]
         org_prefix = org + "/"
         org_prefix_len = len(org_prefix)

orgs/org_management/org_user_management.py

@@ -3,7 +3,7 @@
 import os
 import uuid
 from pathlib import Path
-from typing import final
+from typing import Any, final
 
 import requests
 import yaml
@@ -30,17 +30,17 @@ def __init__(
     def _get_request_headrs(self):
         return {"Authorization": f"Bearer {self.github_token}"}
 
-    def _process_request_result(self, request):
+    def _process_request_result(self, request: requests.Response):
         if request.status_code == 200 or request.status_code == 201:
             return request.json()
         else:
             raise Exception(f"Request execution failed with status code of {request.status_code}. {request.status_code}")
 
-    def _execute_query(self, query):
+    def _execute_query(self, query: str):
         request = requests.post("https://api.github.com/graphql", json={"query": query}, headers=self._get_request_headrs())
         return self._process_request_result(request)
 
-    def _build_query(self, after_cursor_value=None):
+    def _build_query(self, after_cursor_value: Any):
         after_cursor = f'"{after_cursor_value}"' if after_cursor_value else "null"
         query = (
             """
@@ -70,10 +70,10 @@ def _build_query(self, after_cursor_value=None):
         )
         return query
 
-    def get_inactive_users(self):
-        inactive_users = set()
+    def get_inactive_users(self) -> set[str]:
+        inactive_users = set[str]()
         has_next_page = True
-        after_cursor_value = None
+        after_cursor_value: Any = None
         while has_next_page:
             result = self._execute_query(self._build_query(after_cursor_value))
             for user_node in result["data"]["organization"]["membersWithRole"]["nodes"]:
@@ -89,31 +89,31 @@ def get_inactive_users(self):
 
         return inactive_users
 
-    def _load_yaml_file(self, path):
+    def _load_yaml_file(self, path: Path) -> dict[str, Any]:
         with open(path) as stream:
             return yaml.safe_load(stream)
 
-    def _write_yaml_file(self, path, data):
+    def _write_yaml_file(self, path: Path, data: dict[str, Any]):
         with open(path, "w") as f:
             yaml.dump(data, f)
 
-    def _get_inactive_users_msg_for_wgs(self, inactive_users_by_wg, user_tagging_prefix):
+    def _get_inactive_users_msg_for_wgs(self, inactive_users_by_wg: dict[str, set[str]], user_tagging_prefix: str) -> str:
         result = "\n\nWarning:\n" if inactive_users_by_wg else ""
         for wg, users in inactive_users_by_wg.items():
             wg_users_as_list = "\n".join(str(user_tagging_prefix + s) for s in users)
             result += f'Inactive users of Working Group "{wg}" are: \n{wg_users_as_list}\n'
         return result
 
-    def delete_inactive_contributors(self, users_to_delete):
-        path = f"{_SCRIPT_PATH}/contributors.yml"
+    def delete_inactive_contributors(self, users_to_delete: set[str]):
+        path = _SCRIPT_PATH / "contributors.yml"
         contributors_yaml = self._load_yaml_file(path)
-        users_to_delete_lower = [user.lower() for user in users_to_delete]
+        users_to_delete_lower = {user.lower() for user in users_to_delete}
         contributors_yaml["orgs"][self.github_org]["contributors"] = [
             c for c in contributors_yaml["orgs"][self.github_org]["contributors"] if c.lower() not in users_to_delete_lower
         ]
         self._write_yaml_file(path, contributors_yaml)
 
-    def get_inactive_users_msg(self, users_to_delete, inactive_users_by_wg, tagusers):
+    def get_inactive_users_msg(self, users_to_delete: set[str], inactive_users_by_wg: dict[str, set[str]], tagusers: bool) -> str:
         rfc = (
             "https://github.com/cloudfoundry/community/blob/main/toc/rfc/"
             "rfc-0025-define-criteria-and-removal-process-for-inactive-members.md"
@@ -137,17 +137,19 @@ def get_inactive_users_msg(self, users_to_delete, inactive_users_by_wg, tagusers
             f"{self._get_inactive_users_msg_for_wgs(inactive_users_by_wg, user_tagging_prefix)}"
         )
 
-    def get_inactive_users_by_wg(self, inactive_users, community_members_with_role_by_wg):
-        result = dict()
+    def get_inactive_users_by_wg(
+        self, inactive_users: set[str], community_members_with_role_by_wg: dict[str, set[str]]
+    ) -> dict[str, set[str]]:
+        result = dict[str, set[str]]()
         for wg, members in community_members_with_role_by_wg.items():
             wg_inactive_members = inactive_users.intersection(members)
             if len(wg_inactive_members) != 0 and wg != "Admin":
                 result[wg] = wg_inactive_members
         return result
 
     @staticmethod
-    def _get_bool_env_var(env_var_name, default):
-        return os.getenv(env_var_name, default).lower() == "true"
+    def get_bool_env_var(env_var_name: str, default: bool) -> bool:
+        return os.getenv(env_var_name, str(default)).lower() == "true"
 
 
 if __name__ == "__main__":
@@ -178,7 +180,7 @@ def _get_bool_env_var(env_var_name, default):
     generator = OrgGenerator()
     generator.load_from_project()
     community_members_with_role_by_wg = generator.get_community_members_with_role_by_wg(args.githuborg)
-    community_members_with_role = set()
+    community_members_with_role = set[str]()
     for members in community_members_with_role_by_wg.values():
         community_members_with_role |= set(members)
 
@@ -188,11 +190,11 @@ def _get_bool_env_var(env_var_name, default):
 
     print(f"Inactive users length is {len(inactive_users)} and inactive users are {inactive_users}")
     users_to_delete = inactive_users - community_members_with_role
-    tagusers = args.tagusers or InactiveUserHandler._get_bool_env_var("INACTIVE_USER_MANAGEMENT_TAG_USERS", "False")
+    tagusers = args.tagusers or InactiveUserHandler.get_bool_env_var("INACTIVE_USER_MANAGEMENT_TAG_USERS", False)
     inactive_users_by_wg = userHandler.get_inactive_users_by_wg(inactive_users, community_members_with_role_by_wg)
     inactive_users_msg = userHandler.get_inactive_users_msg(users_to_delete, inactive_users_by_wg, tagusers)
     print(f"Inactive users by wg are {inactive_users_by_wg}")
-    if args.dryrun or InactiveUserHandler._get_bool_env_var("INACTIVE_USER_MANAGEMENT_DRY_RUN", "False"):
+    if args.dryrun or InactiveUserHandler.get_bool_env_var("INACTIVE_USER_MANAGEMENT_DRY_RUN", False):
         print(f"Dry-run mode.\nInactive_users_msg is: {inactive_users_msg}")
         print(f"Following users will be deleted: {inactive_users}")
     elif users_to_delete:

orgs/org_management/test_org_management.py

@@ -1,10 +1,13 @@
 import unittest
+from typing import final, override
 
 import jsonschema
 import yaml
 
 from .org_management import OrgGenerator
 
+# pyright: reportPrivateUsage=false
+
 org_cfg = """
 ---
 orgs:
@@ -278,7 +281,9 @@
 """
 
 
+@final
 class TestOrgGenerator(unittest.TestCase):
+    @override
     def setUp(self) -> None:
         OrgGenerator._MANAGED_ORGS = ["cloudfoundry"]
 

orgs/pyproject.toml

@@ -74,7 +74,7 @@ include = ["org_management"]
 # By default BasedPyright is very strict, so you almost certainly want to disable
 # some of the rules.
 # First, these turn off warnings about (yes) how you ignore warnings:
-failOnWarnings = false
+# failOnWarnings = false
 enableTypeIgnoreComments = true
 # reportIgnoreCommentWithoutRule = false
 # reportUnnecessaryTypeIgnoreComment = false