Add: typing in pontos/github (#651)

Co-authored-by: Jaspar S <[email protected]>

pontos/github/actions/argparser.py

@@ -18,7 +18,7 @@
 """ Argument parser for pontos-github-actions """
 
 from argparse import ArgumentParser, Namespace
-from typing import List
+from typing import List, Optional
 
 from .cmds import actions_input, actions_output
 
@@ -30,7 +30,7 @@ def split_pairs(value: str):
 
 
 def parse_args(
-    args: List[str] = None,
+    args: Optional[List[str]] = None,
 ) -> Namespace:
     """
     Parsing args for Pontos GitHub Actions

pontos/github/api/artifacts.py

@@ -26,7 +26,9 @@ class GitHubAsyncRESTArtifacts(GitHubAsyncREST):
     def _get_paged_artifacts(
         self, api, *, params: Optional[Params] = None
     ) -> AsyncIterator[Artifact]:
-        return self._get_paged_items(api, "artifacts", Artifact, params=params)
+        return self._get_paged_items(
+            api, "artifacts", Artifact, params=params  # type: ignore
+        )
 
     def get_all(self, repo: str) -> AsyncIterator[Artifact]:
         """

pontos/github/api/branch.py

@@ -54,7 +54,7 @@ def update_from_applied_settings(
     Return keyword arguments for update_protection_rules by merging existing
     settings with desired updated values.
     """
-    kwargs = {}
+    kwargs: Dict[str, Any] = {}
     if enforce_admins is not None:
         kwargs["enforce_admins"] = enforce_admins
     elif branch_protection.enforce_admins:
@@ -422,7 +422,7 @@ async def update_protection_rules(
             HTTPStatusError if the request was invalid
         """
         api = f"/repos/{repo}/branches/{branch}/protection"
-        data = {
+        data: Dict[str, Any] = {
             "enforce_admins": None,
             "required_status_checks": None,
             "required_pull_request_reviews": None,
@@ -714,7 +714,7 @@ async def update_required_status_checks(
         api = (
             f"/repos/{repo}/branches/{branch}/protection/required_status_checks"
         )
-        data = {}
+        data: Dict[str, Any] = {}
         if require_branches_to_be_up_to_date is not None:
             data["strict"] = require_branches_to_be_up_to_date
         if required_status_checks is not None:

pontos/github/api/client.py

@@ -76,7 +76,7 @@ def _request_headers(
         """
         Get the default request headers
         """
-        headers = {
+        headers: Dict[str, str] = {
             "Accept": accept or DEFAULT_ACCEPT_HEADER,
             "X-GitHub-Api-Version": GITHUB_API_VERSION,
         }
@@ -91,7 +91,7 @@ def _request_headers(
 
     def _request_kwargs(
         self, *, json: Optional[JSON] = None, content: Optional[Any] = None
-    ) -> Dict[str, str]:
+    ) -> JSON:
         """
         Get the default request arguments
         """
@@ -100,7 +100,7 @@ def _request_kwargs(
             kwargs["json"] = json
         if content is not None:
             kwargs["content"] = content
-        return kwargs
+        return kwargs  # type: ignore
 
     def _request_api_url(self, api: str) -> str:
         return f"{self.url}{api}"
@@ -128,7 +128,7 @@ async def get(
         url = self._request_url(api)
         headers = self._request_headers()
         kwargs = self._request_kwargs()
-        return await self._client.get(
+        return await self._client.get(  # type: ignore
             url,
             headers=headers,
             params=params,
@@ -278,7 +278,9 @@ async def __aexit__(
         exc_value: Optional[BaseException],
         traceback: Optional[TracebackType],
     ) -> Optional[bool]:
-        return await self._client.__aexit__(exc_type, exc_value, traceback)
+        return await self._client.__aexit__(  # type: ignore
+            exc_type, exc_value, traceback
+        )
 
 
 class GitHubAsyncREST:
@@ -309,5 +311,5 @@ async def _get_paged_items(
         async for response in self._client.get_all(api, params=params):
             response.raise_for_status()
             data: JSON_OBJECT = response.json()
-            for item in data.get(name, []):
-                yield model_cls.from_dict(item)
+            for item in data.get(name, []):  # type: ignore
+                yield model_cls.from_dict(item)  # type:ignore

pontos/github/api/labels.py

@@ -46,7 +46,7 @@ async def get_all(
             data: JSON = response.json()
 
             for label in data:
-                yield label["name"]
+                yield label["name"]  # type: ignore
 
     async def set_all(
         self, repo: str, issue: int, labels: Iterable[str]
@@ -61,6 +61,6 @@ async def set_all(
                 be overwritten.
         """
         api = f"/repos/{repo}/issues/{issue}/labels"
-        data = {"labels": labels}
+        data: JSON = {"labels": labels}  # type: ignore
         response = await self._client.post(api, data=data)
         response.raise_for_status()

pontos/github/api/organizations.py

@@ -107,7 +107,7 @@ async def invite(
         email: Optional[str] = None,
         invitee_id: Optional[Union[str, int]] = None,
         role: Union[InvitationRole, str] = InvitationRole.DIRECT_MEMBER,
-        team_ids: Iterable[Union[str, int]] = None,
+        team_ids: Optional[Iterable[Union[str, int]]] = None,
     ) -> None:
         """
         Invite a user to a GitHub Organization

pontos/github/api/pull_requests.py

@@ -17,10 +17,10 @@
 
 from collections import defaultdict
 from pathlib import Path
-from typing import AsyncIterator, Dict, Iterable, Optional
+from typing import AsyncIterator, Dict, Iterable, List, Optional
 
 from pontos.github.api.client import GitHubAsyncREST
-from pontos.github.api.helper import FileStatus
+from pontos.github.api.helper import JSON_OBJECT, FileStatus
 from pontos.github.models.pull_request import PullRequest, PullRequestCommit
 
 
@@ -109,7 +109,7 @@ async def create(
             httpx.HTTPStatusError if the request was invalid
         """
         api = f"/repos/{repo}/pulls"
-        data = {
+        data: JSON_OBJECT = {
             "head": head_branch,
             "base": base_branch,
             "title": title,
@@ -147,7 +147,7 @@ async def update(
         """
         api = f"/repos/{repo}/pulls/{pull_request}"
 
-        data = {}
+        data: JSON_OBJECT = {}
         if base_branch:
             data["base"] = base_branch
         if title:
@@ -174,7 +174,7 @@ async def add_comment(
             httpx.HTTPStatusError if the request was invalid
         """
         api = f"/repos/{repo}/issues/{pull_request}/comments"
-        data = {"body": comment}
+        data: JSON_OBJECT = {"body": comment}
         response = await self._client.post(api, data=data)
         response.raise_for_status()
 
@@ -205,7 +205,7 @@ async def files(
         # possible to receive 100
         params = {"per_page": "100"}
         api = f"/repos/{repo}/pulls/{pull_request}/files"
-        file_dict = defaultdict(list)
+        file_dict: Dict[FileStatus, List[Path]] = defaultdict(list)
 
         async for response in self._client.get_all(api, params=params):
             for f in response.json():
@@ -218,4 +218,4 @@ async def files(
                 if not status_list or status in status_list:
                     file_dict[status].append(Path(f["filename"]))
 
-        return file_dict
+        return file_dict  # type: ignore

pontos/github/api/release.py

@@ -29,6 +29,7 @@
 import httpx
 
 from pontos.github.api.client import GitHubAsyncREST
+from pontos.github.api.helper import JSON_OBJECT
 from pontos.github.models.release import Release
 from pontos.helper import AsyncDownloadProgressIterable, download_async, upload
 
@@ -42,8 +43,8 @@ async def create(
         body: Optional[str] = None,
         name: Optional[str] = None,
         target_commitish: Optional[str] = None,
-        draft: Optional[bool] = False,
-        prerelease: Optional[bool] = False,
+        draft: bool = False,
+        prerelease: bool = False,
     ) -> Release:
         """
         Create a new GitHub release
@@ -62,7 +63,7 @@ async def create(
         Raises:
             httpx.HTTPStatusError if the request was invalid
         """
-        data = {
+        data: JSON_OBJECT = {
             "tag_name": tag,
             "draft": draft,
             "prerelease": prerelease,
@@ -284,13 +285,13 @@ async def upload_file(
                 params={"name": file_path.name},
                 content_type=content_type,
                 content_length=file_path.stat().st_size,
-                content=upload(file_path),
+                content=upload(file_path),  # type: ignore
             )
             return response, file_path
 
         tasks = []
         for file_path in files:
-            if isinstance(file_path, Tuple):
+            if isinstance(file_path, tuple):
                 file_path, content_type = file_path
             else:
                 content_type = "application/octet-stream"

pontos/github/api/repositories.py

@@ -18,6 +18,7 @@
 from typing import Optional, Union
 
 from pontos.github.api.client import GitHubAsyncREST
+from pontos.github.api.helper import JSON_OBJECT
 from pontos.github.models.organization import (
     GitIgnoreTemplate,
     LicenseType,
@@ -81,7 +82,7 @@ async def create(
         has_downloads: Optional[bool] = True,
         is_template: Optional[bool] = False,
         team_id: Optional[str] = None,
-        auto_init: Optional[str] = False,
+        auto_init: Optional[bool] = False,
         gitignore_template: Optional[Union[GitIgnoreTemplate, str]] = None,
         license_template: Optional[Union[LicenseType, str]] = None,
         allow_squash_merge: Optional[bool] = True,
@@ -187,7 +188,7 @@ async def create(
                 failed.
         """
         api = f"/orgs/{organization}/repos"
-        data = {"name": name}
+        data: JSON_OBJECT = {"name": name}
 
         if description:
             data["description"] = description
@@ -261,7 +262,7 @@ async def archive(self, repo: str) -> None:
         """
         api = f"/repos/{repo}"
 
-        data = {"archived": True}
+        data: JSON_OBJECT = {"archived": True}
         response = await self._client.post(api, data=data)
         response.raise_for_status()
 
@@ -377,7 +378,7 @@ async def update(
         """
         api = f"/repos/{repo}"
 
-        data = {}
+        data: JSON_OBJECT = {}
         if name:
             data["name"] = name
         if description:

pontos/github/api/tags.py

@@ -16,7 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from datetime import datetime
-from typing import Optional, Union
+from typing import Any, Dict, Optional, Union
 
 from pontos.github.api.client import GitHubAsyncREST
 from pontos.github.models.tag import GitObjectType, Tag
@@ -91,7 +91,7 @@ async def create_tag_reference(
             sha: The SHA1 value for this Github tag.
         """
 
-        data = {
+        data: Dict[str, Any] = {
             "ref": f"refs/tags/{tag}",
             "sha": sha,
         }

pontos/github/api/teams.py

@@ -15,7 +15,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from typing import AsyncIterator, Iterable, Optional
+from typing import Any, AsyncIterator, Dict, Iterable, Optional
 
 from pontos.github.api.client import GitHubAsyncREST
 from pontos.github.models.base import (
@@ -93,7 +93,7 @@ async def create(
             `httpx.HTTPStatusError` if there was an error in the request
         """
         api = f"/orgs/{organization}/teams"
-        data = {"name": name}
+        data: Dict[str, Any] = {"name": name}
         if description:
             data["description"] = description
         if maintainers:
@@ -169,7 +169,7 @@ async def update(
             `httpx.HTTPStatusError` if there was an error in the request
         """
         api = f"/orgs/{organization}/teams/{team}"
-        data = {}
+        data: Dict[str, Any] = {}
         if name:
             data["name"] = name
         if description:
@@ -257,7 +257,7 @@ async def update_member(
             `httpx.HTTPStatusError` if there was an error in the request
         """
         api = f"/orgs/{organization}/teams/{team}/memberships/{username}"
-        data = {"role": role.value}
+        data: Dict[str, Any] = {"role": role.value}
         response = await self._client.put(api, data=data)
         response.raise_for_status()
 
@@ -339,7 +339,7 @@ async def update_permission(
             f"/orgs/{organization}/teams/{team}/repos/{organization}/"
             f"{repository}"
         )
-        data = {"permission": permission.value}
+        data: Dict[str, Any] = {"permission": permission.value}
         response = await self._client.put(api, data=data)
         response.raise_for_status()
 

pontos/github/api/workflows.py

@@ -15,7 +15,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from typing import AsyncIterator, Dict, Optional, Union
+from typing import Any, AsyncIterator, Dict, Optional, Union
 
 from pontos.github.api.client import GitHubAsyncREST
 from pontos.github.models.base import Event
@@ -45,7 +45,7 @@ def get_all(self, repo: str) -> AsyncIterator[Workflow]:
             Information about the workflows as an iterable of dicts
         """
         api = f"/repos/{repo}/actions/workflows"
-        return self._get_paged_items(api, "workflows", Workflow)
+        return self._get_paged_items(api, "workflows", Workflow)  # type: ignore
 
     async def get(self, repo: str, workflow: Union[str, int]) -> Workflow:
         """
@@ -76,7 +76,7 @@ async def create_workflow_dispatch(
         workflow: Union[str, int],
         *,
         ref: str,
-        inputs: Dict[str, str] = None,
+        inputs: Optional[Dict[str, str]] = None,
     ) -> None:
         """
         Create a workflow dispatch event to manually trigger a GitHub Actions
@@ -106,7 +106,7 @@ async def create_workflow_dispatch(
                 )
         """
         api = f"/repos/{repo}/actions/workflows/{workflow}/dispatches"
-        data = {"ref": ref}
+        data: Dict[str, Any] = {"ref": ref}
 
         if inputs:
             data["inputs"] = inputs
@@ -167,7 +167,7 @@ def get_workflow_runs(
             if workflow
             else f"/repos/{repo}/actions/runs"
         )
-        params = {}
+        params: Dict[str, Any] = {}
         if actor:
             params["actor"] = actor
         if branch:
@@ -181,7 +181,7 @@ def get_workflow_runs(
         if exclude_pull_requests is not None:
             params["exclude_pull_requests"] = exclude_pull_requests
 
-        return self._get_paged_items(
+        return self._get_paged_items(  # type: ignore
             api, "workflow_runs", WorkflowRun, params=params
         )