API Improvements: Fix typings for various files pt5

Author: JerrySentry

api/internal/feature/views.py

@@ -1,8 +1,9 @@
 import logging
 import pickle
+from typing import Any, Dict, List
 
 from rest_framework import status
-from rest_framework.response import Response
+from rest_framework.response import Request, Response
 from rest_framework.views import APIView
 from shared.django_apps.rollouts.models import FeatureFlag
 from shared.rollouts import Feature
@@ -20,15 +21,15 @@ class FeaturesView(APIView):
     skip_feature_cache = get_config("setup", "skip_feature_cache", default=False)
     timeout = 300
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.redis = get_redis_connection()
         super().__init__(*args, **kwargs)
 
-    def get_many_from_redis(self, keys):
+    def get_many_from_redis(self, keys: List) -> Dict[str, Any]:
         ret = self.redis.mget(keys)
         return {k: pickle.loads(v) for k, v in zip(keys, ret) if v is not None}
 
-    def set_many_to_redis(self, data):
+    def set_many_to_redis(self, data: Dict[str, Any]) -> None:
         pipeline = self.redis.pipeline()
         pipeline.mset({k: pickle.dumps(v) for k, v in data.items()})
 
@@ -38,7 +39,7 @@ def set_many_to_redis(self, data):
             pipeline.expire(key, self.timeout)
         pipeline.execute()
 
-    def post(self, request):
+    def post(self, request: Request) -> Response:
         serializer = FeatureRequestSerializer(data=request.data)
         if serializer.is_valid():
             flag_evaluations = {}

codecov_auth/management/commands/set_trial_status_values.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from typing import Any
 
 from django.core.management.base import BaseCommand, CommandParser
 from django.db.models import Q
@@ -20,7 +21,7 @@ class Command(BaseCommand):
     def add_arguments(self, parser: CommandParser) -> None:
         parser.add_argument("trial_status_type", type=str)
 
-    def handle(self, *args, **options) -> None:
+    def handle(self, *args: Any, **options: Any) -> None:
         trial_status_type = options.get("trial_status_type", {})
 
         # NOT_STARTED

core/commands/pull/interactors/fetch_pull_request.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timedelta
 
-from shared.django_apps.core.models import Pull
+from shared.django_apps.core.models import Pull, Repository
 
 from codecov.commands.base import BaseInteractor
 from codecov.db import sync_to_async
@@ -17,7 +17,7 @@ def _should_sync_pull(self, pull: Pull | None) -> bool:
         )
 
     @sync_to_async
-    def execute(self, repository, id):
+    def execute(self, repository: Repository, id: int) -> Pull:
         pull = repository.pull_requests.filter(pullid=id).first()
         if self._should_sync_pull(pull):
             TaskService().pulls_sync(repository.repoid, id)

core/commands/repository/interactors/erase_repository.py

@@ -11,7 +11,7 @@
 
 
 class EraseRepositoryInteractor(BaseInteractor):
-    def validate_owner(self, owner: Owner):
+    def validate_owner(self, owner: Owner) -> None:
         if not current_user_part_of_org(self.current_owner, owner):
             raise Unauthorized()
 
@@ -23,7 +23,7 @@ def validate_owner(self, owner: Owner):
                 raise Unauthorized()
 
     @sync_to_async
-    def execute(self, repo_name: str, owner: Owner):
+    def execute(self, repo_name: str, owner: Owner) -> None:
         self.validate_owner(owner)
         repo = Repository.objects.filter(author_id=owner.pk, name=repo_name).first()
         if not repo:

graphql_api/actions/flags.py

@@ -10,7 +10,7 @@
 from timeseries.models import Interval, MeasurementName
 
 
-def flags_for_repo(repository: Repository, filters: Mapping = None) -> QuerySet:
+def flags_for_repo(repository: Repository, filters: Mapping = {}) -> QuerySet:
     queryset = RepositoryFlag.objects.filter(
         repository=repository,
         deleted__isnot=True,

graphql_api/types/branch/branch.py

@@ -1,6 +1,7 @@
 from typing import Optional
 
 from ariadne import ObjectType
+from graphql import GraphQLResolveInfo
 
 from core.models import Branch, Commit
 from graphql_api.dataloader.commit import CommitLoader
@@ -9,13 +10,15 @@
 
 
 @branch_bindable.field("headSha")
-def resolve_head_sha(branch: Branch, info) -> str:
+def resolve_head_sha(branch: Branch, info: GraphQLResolveInfo) -> str:
     head = branch.head
     return head
 
 
 @branch_bindable.field("head")
-async def resolve_head_commit(branch: Branch, info) -> Optional[Commit]:
+async def resolve_head_commit(
+    branch: Branch, info: GraphQLResolveInfo
+) -> Optional[Commit]:
     head = branch.head
     if head:
         loader = CommitLoader.loader(info, branch.repository_id)

graphql_api/types/mutation/save_okta_config/save_okta_config.py

@@ -1,4 +1,7 @@
+from typing import Any, Dict
+
 from ariadne import UnionType
+from graphql import GraphQLResolveInfo
 
 from graphql_api.helpers.mutation import (
     require_authenticated,
@@ -9,7 +12,9 @@
 
 @wrap_error_handling_mutation
 @require_authenticated
-async def resolve_save_okta_config(_, info, input):
+async def resolve_save_okta_config(
+    _: Any, info: GraphQLResolveInfo, input: Dict[str, Any]
+) -> None:
     command = info.context["executor"].get_command("owner")
     return await command.save_okta_config(input)
 

graphs/mixins.py

@@ -1,10 +1,12 @@
+from typing import Any
+
 from django.http import HttpResponse
 from rest_framework import status
-from rest_framework.response import Response
+from rest_framework.response import Request, Response
 
 
 class GraphBadgeAPIMixin(object):
-    def get(self, request, *args, **kwargs):
+    def get(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         ext = self.kwargs.get("ext")
         if ext not in self.extensions:
             return Response(