API Improvements: Fix typings for various files pt1

Author: JerrySentry

billing/views.py

@@ -1,6 +1,6 @@
 import logging
 from datetime import datetime
-from typing import List
+from typing import Any, List
 
 import stripe
 from django.conf import settings
@@ -410,7 +410,7 @@ def checkout_session_completed(
 
         self._log_updated([owner])
 
-    def post(self, request: HttpRequest, *args, **kwargs) -> Response:
+    def post(self, request: HttpRequest, *args: Any, **kwargs: Any) -> Response:
         if settings.STRIPE_ENDPOINT_SECRET is None:
             log.critical(
                 "Stripe endpoint secret improperly configured -- webhooks will not be processed."

codecov_auth/authentication/repo_auth.py

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import List
+from typing import Any, Dict, List, Optional, Tuple
 from uuid import UUID
 
 from django.core.exceptions import ObjectDoesNotExist
@@ -10,6 +10,7 @@
 from jwt import PyJWTError
 from rest_framework import authentication, exceptions, serializers
 from rest_framework.exceptions import NotAuthenticated
+from rest_framework.response import Response
 from rest_framework.views import exception_handler
 from shared.django_apps.codecov_auth.models import Owner
 
@@ -32,7 +33,9 @@
 log = logging.getLogger(__name__)
 
 
-def repo_auth_custom_exception_handler(exc, context):
+def repo_auth_custom_exception_handler(
+    exc: Exception, context: Dict[str, Any]
+) -> Response:
     """
     User arrives here if they have correctly supplied a Token or the Tokenless Headers,
     but their Token has not matched with any of our Authentication methods. The goal is to
@@ -60,17 +63,17 @@ def repo_auth_custom_exception_handler(exc, context):
 
 
 class LegacyTokenRepositoryAuth(RepositoryAuthInterface):
-    def __init__(self, repository, auth_data):
+    def __init__(self, repository: Repository, auth_data: Dict[str, Any]) -> None:
         self._auth_data = auth_data
         self._repository = repository
 
-    def get_scopes(self):
+    def get_scopes(self) -> List[TokenTypeChoices]:
         return [TokenTypeChoices.UPLOAD]
 
-    def get_repositories(self):
+    def get_repositories(self) -> List[Repository]:
         return [self._repository]
 
-    def allows_repo(self, repository):
+    def allows_repo(self, repository: Repository) -> bool:
         return repository in self.get_repositories()
 
 
@@ -79,17 +82,17 @@ class OIDCTokenRepositoryAuth(LegacyTokenRepositoryAuth):
 
 
 class TableTokenRepositoryAuth(RepositoryAuthInterface):
-    def __init__(self, repository, token):
+    def __init__(self, repository: Repository, token: RepositoryToken) -> None:
         self._token = token
         self._repository = repository
 
-    def get_scopes(self):
+    def get_scopes(self) -> List[str]:
         return [self._token.token_type]
 
-    def get_repositories(self):
+    def get_repositories(self) -> List[Repository]:
         return [self._repository]
 
-    def allows_repo(self, repository):
+    def allows_repo(self, repository: Repository) -> bool:
         return repository in self.get_repositories()
 
 
@@ -98,10 +101,10 @@ def __init__(self, token: OrganizationLevelToken) -> None:
         self._token = token
         self._org = token.owner
 
-    def get_scopes(self):
+    def get_scopes(self) -> List[str]:
         return [self._token.token_type]
 
-    def allows_repo(self, repository):
+    def allows_repo(self, repository: Repository) -> bool:
         return repository.author.ownerid == self._org.ownerid
 
     def get_repositories_queryset(self) -> QuerySet:
@@ -120,18 +123,20 @@ class TokenlessAuth(RepositoryAuthInterface):
     def __init__(self, repository: Repository) -> None:
         self._repository = repository
 
-    def get_scopes(self):
+    def get_scopes(self) -> List[TokenTypeChoices]:
         return [TokenTypeChoices.UPLOAD]
 
-    def allows_repo(self, repository):
+    def allows_repo(self, repository: Repository) -> bool:
         return repository in self.get_repositories()
 
     def get_repositories(self) -> List[Repository]:
         return [self._repository]
 
 
 class RepositoryLegacyQueryTokenAuthentication(authentication.BaseAuthentication):
-    def authenticate(self, request):
+    def authenticate(
+        self, request: HttpRequest
+    ) -> Optional[Tuple[RepositoryAsUser, LegacyTokenRepositoryAuth]]:
         token = request.GET.get("token")
         if not token:
             return None
@@ -150,22 +155,26 @@ def authenticate(self, request):
 
 
 class RepositoryLegacyTokenAuthentication(authentication.TokenAuthentication):
-    def authenticate_credentials(self, token):
+    def authenticate_credentials(
+        self, token: str
+    ) -> Optional[Tuple[RepositoryAsUser, LegacyTokenRepositoryAuth]]:
         try:
-            token = UUID(token)
-            repository = Repository.objects.get(upload_token=token)
+            token_uuid = UUID(token)
+            repository = Repository.objects.get(upload_token=token_uuid)
         except (ValueError, TypeError, Repository.DoesNotExist):
             return None  # continue to next auth class
         return (
             RepositoryAsUser(repository),
-            LegacyTokenRepositoryAuth(repository, {"token": token}),
+            LegacyTokenRepositoryAuth(repository, {"token": token_uuid}),
         )
 
 
 class RepositoryTokenAuthentication(authentication.TokenAuthentication):
     keyword = "Repotoken"
 
-    def authenticate_credentials(self, key):
+    def authenticate_credentials(
+        self, key: str
+    ) -> Optional[Tuple[RepositoryAsUser, TableTokenRepositoryAuth]]:
         try:
             token = RepositoryToken.objects.select_related("repository").get(key=key)
         except RepositoryToken.DoesNotExist:
@@ -182,7 +191,9 @@ def authenticate_credentials(self, key):
 
 
 class GlobalTokenAuthentication(authentication.TokenAuthentication):
-    def authenticate(self, request):
+    def authenticate(
+        self, request: HttpRequest
+    ) -> Optional[Tuple[RepositoryAsUser, LegacyTokenRepositoryAuth]]:
         global_tokens = get_global_tokens()
         token = self.get_token(request)
         using_global_token = token in global_tokens
@@ -219,7 +230,9 @@ def get_token(self, request: HttpRequest) -> str | None:
 
 
 class OrgLevelTokenAuthentication(authentication.TokenAuthentication):
-    def authenticate_credentials(self, key):
+    def authenticate_credentials(
+        self, key: str
+    ) -> Optional[Tuple[Owner, OrgLevelTokenRepositoryAuth]]:
         if is_uuid(key):  # else, continue to next auth class
             # Actual verification for org level tokens
             token = OrganizationLevelToken.objects.filter(token=key).first()
@@ -236,7 +249,9 @@ def authenticate_credentials(self, key):
 
 
 class GitHubOIDCTokenAuthentication(authentication.TokenAuthentication):
-    def authenticate_credentials(self, token):
+    def authenticate_credentials(
+        self, token: str
+    ) -> Optional[Tuple[RepositoryAsUser, OIDCTokenRepositoryAuth]]:
         if not token or is_uuid(token):
             return None  # continue to next auth class
 
@@ -283,7 +298,12 @@ def _get_info_from_request_path(
 
         return repo, commitid
 
-    def get_branch(self, request, repoid=None, commitid=None):
+    def get_branch(
+        self,
+        request: HttpRequest,
+        repoid: Optional[int] = None,
+        commitid: Optional[str] = None,
+    ) -> Optional[str]:
         if repoid and commitid:
             commit = Commit.objects.filter(
                 repository_id=repoid, commitid=commitid
@@ -299,7 +319,9 @@ def get_branch(self, request, repoid=None, commitid=None):
             else:
                 return body.get("branch")
 
-    def authenticate(self, request):
+    def authenticate(
+        self, request: HttpRequest
+    ) -> Tuple[RepositoryAsUser, TokenlessAuth]:
         repository, commitid = self._get_info_from_request_path(request)
 
         if repository is None or repository.private:
@@ -341,7 +363,12 @@ def _get_info_from_request_path(
             # Validate provider
             raise exceptions.AuthenticationFailed(self.auth_failed_message)
 
-    def get_branch(self, request, repoid=None, commitid=None):
+    def get_branch(
+        self,
+        request: HttpRequest,
+        repoid: Optional[int] = None,
+        commitid: Optional[str] = None,
+    ) -> str:
         body = json.loads(str(request.body, "utf8"))
 
         # If commit is not created yet (ie first upload for this commit), we just validate branch format.
@@ -419,7 +446,7 @@ class UploadTokenRequiredGetFromBodyAuthenticationCheck(
     then use the same authenticate() as parent class.
     """
 
-    def _get_git(self, validated_data):
+    def _get_git(self, validated_data: Dict[str, str]) -> Optional[str]:
         """
         BA sends this in as git_service, TA sends this in as service.
         Use this function so this Check class can be used by both views.

graphql_api/types/commit/commit.py

@@ -84,7 +84,7 @@ async def resolve_yaml(commit: Commit, info) -> dict:
 
 
 @commit_bindable.field("yamlState")
-async def resolve_yaml_state(commit: Commit, info) -> YamlStates:
+async def resolve_yaml_state(commit: Commit, info) -> Optional[YamlStates]:
     command = info.context["executor"].get_command("commit")
     final_yaml = await command.get_final_yaml(commit)
     return get_yaml_state(yaml=final_yaml)

graphql_api/types/owner/owner.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from hashlib import sha1
-from typing import Any, Iterable, List, Optional
+from typing import Any, Coroutine, Iterable, List, Optional
 
 import shared.rate_limits as rate_limits
 import stripe
@@ -62,7 +62,7 @@ def resolve_repositories(
     ordering: Optional[RepositoryOrdering] = RepositoryOrdering.ID,
     ordering_direction: Optional[OrderingDirection] = OrderingDirection.ASC,
     **kwargs: Any,
-) -> Connection:
+) -> Coroutine[Any, Any, Connection]:
     current_owner = info.context["request"].current_owner
     okta_account_auths: list[int] = info.context["request"].session.get(
         OKTA_SIGNED_IN_ACCOUNTS_SESSION_KEY, []

services/yaml.py

@@ -1,7 +1,7 @@
 import enum
 import logging
 from functools import lru_cache
-from typing import Dict
+from typing import Dict, Optional
 
 from asgiref.sync import async_to_sync
 from shared.yaml import UserYaml, fetch_current_yaml_from_provider_via_reference
@@ -70,6 +70,6 @@ def final_commit_yaml(commit: Commit, owner: Owner | None) -> UserYaml:
     )
 
 
-def get_yaml_state(yaml: UserYaml) -> YamlStates:
+def get_yaml_state(yaml: UserYaml) -> Optional[YamlStates]:
     if yaml == get_config("site", default={}):
         return YamlStates.DEFAULT

upload/views/commits.py

@@ -1,7 +1,12 @@
 import logging
+from typing import Any, Callable, Dict
 
+from django.db.models import QuerySet
+from django.http import HttpRequest
+from rest_framework import serializers
 from rest_framework.exceptions import NotAuthenticated
 from rest_framework.generics import ListCreateAPIView
+from rest_framework.response import Response
 from shared.metrics import inc_counter
 
 from codecov_auth.authentication.repo_auth import (
@@ -13,7 +18,7 @@
     UploadTokenRequiredAuthenticationCheck,
     repo_auth_custom_exception_handler,
 )
-from core.models import Commit
+from core.models import Commit, Repository
 from upload.helpers import (
     generate_upload_prometheus_metrics_labels,
     validate_activated_repo,
@@ -26,7 +31,9 @@
 log = logging.getLogger(__name__)
 
 
-def create_commit(serializer, repository):
+def create_commit(
+    serializer: serializers.ModelSerializer, repository: Repository
+) -> Commit:
     validate_activated_repo(repository)
     commit = serializer.save(repository=repository)
     return commit
@@ -44,25 +51,25 @@ class CommitViews(ListCreateAPIView, GetterMixin):
         TokenlessAuthentication,
     ]
 
-    def get_exception_handler(self):
+    def get_exception_handler(self) -> Callable[[Exception, Dict[str, Any]], Response]:
         return repo_auth_custom_exception_handler
 
-    def get_queryset(self):
+    def get_queryset(self) -> QuerySet:
         repository = self.get_repo()
         return Commit.objects.filter(repository=repository)
 
-    def list(self, request, *args, **kwargs):
+    def list(self, request: HttpRequest, *args: Any, **kwargs: Any) -> Response:
         repository = self.get_repo()
         if repository.private and isinstance(
             self.request.auth, TokenlessAuthentication
         ):
             raise NotAuthenticated()
         return super().list(request, *args, **kwargs)
 
-    def create(self, request, *args, **kwargs):
+    def create(self, request: HttpRequest, *args: Any, **kwargs: Any) -> Response:
         return super().create(request, *args, **kwargs)
 
-    def perform_create(self, serializer):
+    def perform_create(self, serializer: CommitSerializer) -> Commit:
         inc_counter(
             API_UPLOAD_COUNTER,
             labels=generate_upload_prometheus_metrics_labels(