add typing 1

Author: JerrySentry

codecov_auth/commands/owner/interactors/set_yaml_on_owner.py

@@ -1,4 +1,5 @@
 import html
+from typing import Optional
 
 import yaml
 from shared.validation.exceptions import InvalidYamlException
@@ -17,21 +18,21 @@
 
 
 class SetYamlOnOwnerInteractor(BaseInteractor):
-    def validate(self):
+    def validate(self) -> None:
         if not self.current_user.is_authenticated:
             raise Unauthenticated()
 
-    def authorize(self):
+    def authorize(self) -> None:
         if not current_user_part_of_org(self.current_owner, self.owner):
             raise Unauthorized()
 
-    def get_owner(self, username):
+    def get_owner(self, username: str) -> Owner:
         try:
             return Owner.objects.get(username=username, service=self.service)
         except Owner.DoesNotExist:
             raise NotFound()
 
-    def convert_yaml_to_dict(self, yaml_input):
+    def convert_yaml_to_dict(self, yaml_input: str) -> Optional[dict]:
         yaml_safe = html.escape(yaml_input, quote=False)
         try:
             yaml_dict = yaml.safe_load(yaml_safe)
@@ -49,7 +50,7 @@ def convert_yaml_to_dict(self, yaml_input):
             raise ValidationError(message)
 
     @sync_to_async
-    def execute(self, username, yaml_input):
+    def execute(self, username: str, yaml_input: str) -> Owner:
         self.validate()
         self.owner = self.get_owner(username)
         self.authorize()

graphql_api/types/owner/owner.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from hashlib import sha1
-from typing import Iterable, List, Optional
+from typing import Any, Iterable, List, Optional
 
 import shared.rate_limits as rate_limits
 import stripe
@@ -26,6 +26,7 @@
 )
 from graphql_api.helpers.ariadne import ariadne_load_local_graphql
 from graphql_api.helpers.connection import (
+    Connection,
     build_connection_graphql,
     queryset_to_connection,
 )
@@ -38,7 +39,7 @@
 from services.profiling import ProfilingSummary
 from services.redis_configuration import get_redis_connection
 from timeseries.helpers import fill_sparse_measurements
-from timeseries.models import Interval, MeasurementSummary
+from timeseries.models import Interval
 from utils.config import get_config
 
 owner = ariadne_load_local_graphql(__file__, "owner.graphql")
@@ -51,12 +52,12 @@
 @convert_kwargs_to_snake_case
 def resolve_repositories(
     owner: Owner,
-    info,
-    filters=None,
-    ordering=RepositoryOrdering.ID,
-    ordering_direction=OrderingDirection.ASC,
-    **kwargs,
-):
+    info: GraphQLResolveInfo,
+    filters: Optional[dict] = None,
+    ordering: Optional[RepositoryOrdering] = RepositoryOrdering.ID,
+    ordering_direction: Optional[OrderingDirection] = OrderingDirection.ASC,
+    **kwargs: 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, []
@@ -81,18 +82,18 @@ def resolve_repositories(
 
 @owner_bindable.field("isCurrentUserPartOfOrg")
 @sync_to_async
-def resolve_is_current_user_part_of_org(owner, info: GraphQLResolveInfo):
+def resolve_is_current_user_part_of_org(owner: Owner, info: GraphQLResolveInfo) -> bool:
     current_owner = info.context["request"].current_owner
     return current_user_part_of_org(current_owner, owner)
 
 
 @owner_bindable.field("yaml")
-def resolve_yaml(owner: Owner, info: GraphQLResolveInfo):
+def resolve_yaml(owner: Owner, info: GraphQLResolveInfo) -> Optional[str]:
     if owner.yaml is None:
-        return
+        return None
     current_owner = info.context["request"].current_owner
     if not current_user_part_of_org(current_owner, owner):
-        return
+        return None
     return owner.yaml.get("to_string", yaml.dump(owner.yaml))
 
 
@@ -134,7 +135,9 @@ def resolve_ownerid(owner: Owner, info: GraphQLResolveInfo) -> int:
 
 
 @owner_bindable.field("repository")
-async def resolve_repository(owner: Owner, info, name):
+async def resolve_repository(
+    owner: Owner, info: GraphQLResolveInfo, name: str
+) -> Repository | NotFoundError:
     command = info.context["executor"].get_command("repository")
     okta_authenticated_accounts: list[int] = info.context["request"].session.get(
         OKTA_SIGNED_IN_ACCOUNTS_SESSION_KEY, []
@@ -174,37 +177,43 @@ async def resolve_repository(owner: Owner, info, name):
 
 @owner_bindable.field("numberOfUploads")
 @require_part_of_org
-async def resolve_number_of_uploads(owner: Owner, info, **kwargs):
+async def resolve_number_of_uploads(
+    owner: Owner, info: GraphQLResolveInfo, **kwargs: Any
+) -> int:
     command = info.context["executor"].get_command("owner")
     return await command.get_uploads_number_per_user(owner)
 
 
 @owner_bindable.field("isAdmin")
 @require_part_of_org
-def resolve_is_current_user_an_admin(owner: Owner, info: GraphQLResolveInfo):
+def resolve_is_current_user_an_admin(owner: Owner, info: GraphQLResolveInfo) -> bool:
     current_owner = info.context["request"].current_owner
     command = info.context["executor"].get_command("owner")
     return command.get_is_current_user_an_admin(owner, current_owner)
 
 
 @owner_bindable.field("hashOwnerid")
 @require_part_of_org
-def resolve_hash_ownerid(owner: Owner, info: GraphQLResolveInfo):
+def resolve_hash_ownerid(owner: Owner, info: GraphQLResolveInfo) -> str:
     hash_ownerid = sha1(str(owner.ownerid).encode())
     return hash_ownerid.hexdigest()
 
 
 @owner_bindable.field("orgUploadToken")
 @require_part_of_org
-def resolve_org_upload_token(owner: Owner, info, **kwargs):
+def resolve_org_upload_token(
+    owner: Owner, info: GraphQLResolveInfo, **kwargs: Any
+) -> str:
     command = info.context["executor"].get_command("owner")
     return command.get_org_upload_token(owner)
 
 
 @owner_bindable.field("defaultOrgUsername")
 @sync_to_async
 @require_part_of_org
-def resolve_org_default_org_username(owner: Owner, info, **kwargs) -> int:
+def resolve_org_default_org_username(
+    owner: Owner, info: GraphQLResolveInfo, **kwargs: Any
+) -> Optional[str]:
     return None if owner.default_org is None else owner.default_org.username
 
 
@@ -213,13 +222,13 @@ def resolve_org_default_org_username(owner: Owner, info, **kwargs) -> int:
 @convert_kwargs_to_snake_case
 def resolve_measurements(
     owner: Owner,
-    info,
+    info: GraphQLResolveInfo,
     interval: Interval,
     after: Optional[datetime] = None,
     before: Optional[datetime] = None,
     repos: Optional[List[str]] = None,
     is_public: Optional[bool] = None,
-) -> Iterable[MeasurementSummary]:
+) -> Iterable[dict]:
     current_owner = info.context["request"].current_owner
 
     okta_authenticated_accounts: list[int] = info.context["request"].session.get(
@@ -256,7 +265,7 @@ def resolve_measurements(
 
 @owner_bindable.field("isCurrentUserActivated")
 @sync_to_async
-def resolve_is_current_user_activated(owner: Owner, info: GraphQLResolveInfo):
+def resolve_is_current_user_activated(owner: Owner, info: GraphQLResolveInfo) -> bool:
     current_user = info.context["request"].user
     if not current_user.is_authenticated:
         return False
@@ -303,7 +312,7 @@ def resolve_is_github_rate_limited(
 @convert_kwargs_to_snake_case
 def resolve_owner_invoice(
     owner: Owner,
-    info,
+    info: GraphQLResolveInfo,
     invoice_id: str,
 ) -> stripe.Invoice | None:
     return BillingService(requesting_user=owner).get_invoice(owner, invoice_id)
@@ -371,7 +380,9 @@ def resolve_ai_enabled_repos(
 
 @owner_bindable.field("uploadTokenRequired")
 @require_part_of_org
-def resolve_upload_token_required(owner: Owner, info) -> bool | None:
+def resolve_upload_token_required(
+    owner: Owner, info: GraphQLResolveInfo
+) -> bool | None:
     return owner.upload_token_required_for_public_repos