API Improvements: Fix typings for various files pt4 (#1086)

Author: JerrySentry

api/internal/commit/serializers.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Dict, List
 
 import shared.reports.api_report_service as report_service
 from rest_framework import serializers
@@ -32,7 +33,7 @@ class Meta:
 class CommitWithFileLevelReportSerializer(CommitSerializer):
     report = serializers.SerializerMethodField()
 
-    def get_report(self, commit: Commit):
+    def get_report(self, commit: Commit) -> Dict[str, List[Dict] | Dict] | None:
         report = report_service.build_report_from_commit(commit)
         if report is None:
             return None

codecov_auth/commands/owner/interactors/save_terms_agreement.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Optional
+from typing import Any, Optional
 
 from django.utils import timezone
 
@@ -20,7 +20,7 @@ class TermsAgreementInput:
 class SaveTermsAgreementInteractor(BaseInteractor):
     requires_service = False
 
-    def validate(self, input: TermsAgreementInput):
+    def validate(self, input: TermsAgreementInput) -> None:
         valid_customer_intents = ["Business", "BUSINESS", "Personal", "PERSONAL"]
         if (
             input.customer_intent
@@ -30,7 +30,7 @@ def validate(self, input: TermsAgreementInput):
         if not self.current_user.is_authenticated:
             raise Unauthenticated()
 
-    def update_terms_agreement(self, input: TermsAgreementInput):
+    def update_terms_agreement(self, input: TermsAgreementInput) -> None:
         self.current_user.terms_agreement = input.terms_agreement
         self.current_user.terms_agreement_at = timezone.now()
         self.current_user.customer_intent = input.customer_intent
@@ -44,14 +44,14 @@ def update_terms_agreement(self, input: TermsAgreementInput):
         if input.marketing_consent:
             self.send_data_to_marketo()
 
-    def send_data_to_marketo(self):
+    def send_data_to_marketo(self) -> None:
         event_data = {
             "email": self.current_user.email,
         }
         AnalyticsService().opt_in_email(self.current_user.id, event_data)
 
     @sync_to_async
-    def execute(self, input):
+    def execute(self, input: Any) -> None:
         typed_input = TermsAgreementInput(
             business_email=input.get("business_email"),
             terms_agreement=input.get("terms_agreement"),

codecov_auth/signals.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Optional, Type
+from typing import Any, Dict, Optional, Type, cast
 
 from django.db.models.signals import post_save
 from django.dispatch import receiver
@@ -38,7 +38,7 @@ def update_owner(
     """
     Shelter tracks a limited set of Owner fields - only update if those fields have changed.
     """
-    created: bool = kwargs["created"]
+    created: bool = cast(bool, kwargs["created"])
     tracked_fields = [
         "upload_token_required_for_public_repos",
         "username",

codecov_auth/views/sentry.py

@@ -123,7 +123,7 @@ def _perform_login(self, request: HttpRequest) -> HttpResponse:
             # user has not connected any owners yet
             return redirect(f"{settings.CODECOV_DASHBOARD_URL}/sync")
 
-    def _login_user(self, request: HttpRequest, user_data: dict):
+    def _login_user(self, request: HttpRequest, user_data: dict) -> User:
         sentry_id = user_data["user"]["id"]
         user_name = user_data["user"].get("name")
         user_email = user_data["user"].get("email")
@@ -177,7 +177,7 @@ def _login_user(self, request: HttpRequest, user_data: dict):
         login(request, current_user)
         return current_user
 
-    def get(self, request):
+    def get(self, request: HttpRequest) -> HttpResponse:
         if request.GET.get("code"):
             return self._perform_login(request)
         else:

core/commands/commit/interactors/get_file_content.py

@@ -1,13 +1,15 @@
 import logging
+from typing import Any, Coroutine
 
 from codecov.commands.base import BaseInteractor
+from core.models import Commit
 from services.repo_providers import RepoProviderService
 
 log = logging.getLogger(__name__)
 
 
 class GetFileContentInteractor(BaseInteractor):
-    async def get_file_from_service(self, commit, path):
+    async def get_file_from_service(self, commit: Commit, path: str) -> str | None:
         try:
             repository_service = await RepoProviderService().async_get_adapter(
                 owner=self.current_owner, repo=commit.repository
@@ -27,5 +29,5 @@ async def get_file_from_service(self, commit, path):
             )
             return None
 
-    def execute(self, commit, path):
+    def execute(self, commit: Commit, path: str) -> Coroutine[Any, Any, str | None]:
         return self.get_file_from_service(commit, path)

graphql_api/actions/repository.py

@@ -9,7 +9,9 @@
 log = logging.getLogger(__name__)
 
 
-def apply_filters_to_queryset(queryset, filters: dict[str, Any]) -> QuerySet:
+def apply_filters_to_queryset(
+    queryset: QuerySet, filters: dict[str, Any] | None
+) -> QuerySet:
     filters = filters or {}
     term = filters.get("term")
     active = filters.get("active")

upload/serializers.py

@@ -1,4 +1,7 @@
+from typing import Any, Dict, List
+
 from django.conf import settings
+from django.db.models import QuerySet
 from rest_framework import serializers
 from shared.api_archive.archive import ArchiveService
 
@@ -11,7 +14,7 @@
 class FlagListField(serializers.ListField):
     child = serializers.CharField()
 
-    def to_representation(self, data):
+    def to_representation(self, data: QuerySet) -> List[str | None]:
         return [item.flag_name if item is not None else None for item in data.all()]
 
 
@@ -47,12 +50,12 @@ class Meta:
 
     raw_upload_location = serializers.SerializerMethodField()
 
-    def get_raw_upload_location(self, obj: ReportSession):
+    def get_raw_upload_location(self, obj: ReportSession) -> str:
         repo = obj.report.commit.repository
         archive_service = ArchiveService(repo)
         return archive_service.create_presigned_put(obj.storage_path)
 
-    def get_url(self, obj: ReportSession):
+    def get_url(self, obj: ReportSession) -> str:
         repository = obj.report.commit.repository
         commit = obj.report.commit
         return f"{settings.CODECOV_DASHBOARD_URL}/{repository.author.service}/{repository.author.username}/{repository.name}/commit/{commit.commitid}"
@@ -64,7 +67,7 @@ def _create_existing_flags_map(self, repoid: int) -> dict:
             existing_flags_map[flag_obj.flag_name] = flag_obj
         return existing_flags_map
 
-    def create(self, validated_data):
+    def create(self, validated_data: Dict[str, Any]) -> ReportSession | None:
         flag_names = (
             validated_data.pop("flags") if "flags" in validated_data.keys() else []
         )
@@ -134,7 +137,7 @@ class Meta:
             "branch",
         )
 
-    def create(self, validated_data):
+    def create(self, validated_data: Dict[str, Any]) -> Commit:
         repo = validated_data.pop("repository", None)
         commitid = validated_data.pop("commitid", None)
         commit, created = Commit.objects.get_or_create(
@@ -161,7 +164,7 @@ class Meta:
         )
         fields = read_only_fields + ("code",)
 
-    def create(self, validated_data) -> tuple[CommitReport, bool]:
+    def create(self, validated_data: Dict[str, Any]) -> tuple[CommitReport, bool]:
         report = (
             CommitReport.objects.coverage_reports()
             .filter(

upload/tokenless/azure.py

@@ -1,5 +1,6 @@
 import logging
 from datetime import datetime, timedelta
+from typing import Any, Dict
 
 import requests
 from requests.exceptions import ConnectionError, HTTPError
@@ -12,7 +13,7 @@
 
 
 class TokenlessAzureHandler(BaseTokenlessUploadHandler):
-    def get_build(self):
+    def get_build(self) -> Dict[str, Any]:
         try:
             response = requests.get(
                 f"{self.server_uri}{self.project}/_apis/build/builds/{self.job}?api-version=5.0",
@@ -54,7 +55,7 @@ def get_build(self):
             )
         return build
 
-    def verify(self):
+    def verify(self) -> None:
         if not self.upload_params.get("job"):
             raise NotFound(
                 'Missing "job" argument. Please upload with the Codecov repository upload token to resolve issue.'

upload/views/uploads.py

@@ -1,10 +1,13 @@
 import logging
+from typing import Any, Callable, Dict
 
 from django.http import HttpRequest, HttpResponseNotAllowed
 from django.utils import timezone
 from rest_framework.exceptions import ValidationError
 from rest_framework.generics import ListCreateAPIView
 from rest_framework.permissions import BasePermission
+from rest_framework.response import Response
+from rest_framework.views import APIView
 from shared.api_archive.archive import ArchiveService, MinioEndpoints
 from shared.metrics import inc_counter
 from shared.upload.utils import UploaderType, insert_coverage_measurement
@@ -40,8 +43,13 @@
 
 
 def create_upload(
-    serializer, repository, commit, report, is_shelter_request, analytics_token
-):
+    serializer: UploadSerializer,
+    repository: Repository,
+    commit: Commit,
+    report: CommitReport,
+    is_shelter_request: bool,
+    analytics_token: str,
+) -> ReportSession:
     version = (
         serializer.validated_data["version"]
         if "version" in serializer.validated_data
@@ -82,7 +90,9 @@ def create_upload(
     return instance
 
 
-def trigger_upload_task(repository, commit_sha, upload, report):
+def trigger_upload_task(
+    repository: Repository, commit_sha: str, upload: ReportSession, report: CommitReport
+) -> None:
     log.info(
         "Triggering upload task",
         extra=dict(
@@ -103,7 +113,7 @@ def trigger_upload_task(repository, commit_sha, upload, report):
     dispatch_upload_task(task_arguments, repository, redis)
 
 
-def activate_repo(repository):
+def activate_repo(repository: Repository) -> None:
     # Only update the fields if needed
     if (
         repository.activated
@@ -128,8 +138,8 @@ def activate_repo(repository):
 
 
 def send_analytics_data(
-    commit: Commit, upload: ReportSession, version, analytics_token
-):
+    commit: Commit, upload: ReportSession, version: str, analytics_token: str
+) -> None:
     analytics_upload_data = {
         "commit": commit.commitid,
         "branch": commit.branch,
@@ -154,7 +164,7 @@ def send_analytics_data(
     )
 
 
-def get_token_for_analytics(commit: Commit, request):
+def get_token_for_analytics(commit: Commit, request: HttpRequest) -> str:
     repo = commit.repository
     if isinstance(request.auth, TokenlessAuth):
         analytics_token = "tokenless_upload"
@@ -170,7 +180,7 @@ def get_token_for_analytics(commit: Commit, request):
 
 
 class CanDoCoverageUploadsPermission(BasePermission):
-    def has_permission(self, request, view):
+    def has_permission(self, request: HttpRequest, view: APIView) -> bool:
         repository = view.get_repo()
         return (
             request.auth is not None
@@ -194,10 +204,10 @@ class UploadViews(ListCreateAPIView, GetterMixin):
     ]
     throttle_classes = [UploadsPerCommitThrottle, UploadsPerWindowThrottle]
 
-    def get_exception_handler(self):
+    def get_exception_handler(self) -> Callable[[Exception, Dict[str, Any]], Response]:
         return repo_auth_custom_exception_handler
 
-    def perform_create(self, serializer: UploadSerializer):
+    def perform_create(self, serializer: UploadSerializer) -> ReportSession:
         inc_counter(
             API_UPLOAD_COUNTER,
             labels=generate_upload_prometheus_metrics_labels(
@@ -253,7 +263,7 @@ def list(
         repo: str,
         commit_sha: str,
         report_code: str,
-    ):
+    ) -> HttpResponseNotAllowed:
         return HttpResponseNotAllowed(permitted_methods=["POST"])
 
     def get_repo(self) -> Repository:
@@ -270,7 +280,7 @@ def get_commit(self, repo: Repository) -> Commit:
         except ValidationError as excpetion:
             raise excpetion
 
-    def get_report(self, commit: Commit) -> CommitReport:
+    def get_report(self, commit: Commit, _: Any = None) -> CommitReport:
         try:
             report = super().get_report(commit)
             return report

validate/views.py

@@ -1,8 +1,9 @@
 import logging
 from json import dumps
+from typing import Any
 
 from django.conf import settings
-from django.http import HttpResponse
+from django.http import HttpRequest, HttpResponse
 from rest_framework import status
 from rest_framework.permissions import AllowAny
 from rest_framework.response import Response
@@ -23,14 +24,14 @@
 class V1ValidateYamlHandler(APIView):
     permission_classes = [AllowAny]
 
-    def get(self, request, *args, **kwargs):
+    def get(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
         return HttpResponse(
             f"Usage:\n\ncurl -X POST --data-binary @codecov.yml {settings.CODECOV_URL}/validate\n",
             status=status.HTTP_200_OK,
             content_type="text/plain",
         )
 
-    def post(self, request, *args, **kwargs):
+    def post(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
         if not self.request.body:
             return HttpResponse(
                 "No content posted.",
@@ -82,7 +83,7 @@ def post(self, request, *args, **kwargs):
 class V2ValidateYamlHandler(V1ValidateYamlHandler):
     permission_classes = [AllowAny]
 
-    def post(self, request, *args, **kwargs):
+    def post(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
         source = self.request.query_params.get("source", "unknown")
         inc_counter(
             API_VALIDATE_V2_COUNTER,