fix(python-types): Enforce static typing in api/app namespace

api/app/pagination.py

@@ -1,6 +1,7 @@
 import base64
 import json
 from collections import OrderedDict
+from typing import Any, List, Optional, Type
 
 from drf_yasg import openapi  # type: ignore[import-untyped]
 from drf_yasg.inspectors import PaginatorInspector  # type: ignore[import-untyped]
@@ -17,7 +18,7 @@ class CustomPagination(PageNumberPagination):
 
 class EdgeIdentityPaginationInspector(PaginatorInspector):  # type: ignore[misc]
     def get_paginator_parameters(
-        self, paginator: BasePagination
+        self, paginator: Type[BasePagination]
     ) -> list[openapi.Parameter]:
         """
         :param BasePagination paginator: the paginator
@@ -40,13 +41,14 @@ def get_paginator_parameters(
             ),
         ]
 
-    def get_paginated_response(self, paginator, response_schema):  # type: ignore[no-untyped-def]
+    def get_paginated_response(
+        self, paginator: Type[BasePagination], response_schema: openapi.Schema
+    ) -> openapi.Schema:
         """
         :param BasePagination paginator: the paginator
         :param openapi.Schema response_schema: the response schema that must be paged.
         :rtype: openapi.Schema
         """
-
         return openapi.Schema(
             type=openapi.TYPE_OBJECT,
             properties=OrderedDict(
@@ -69,7 +71,9 @@ class EdgeIdentityPagination(CustomPagination):
     max_page_size = 100
     page_size = 100
 
-    def paginate_queryset(self, dynamo_queryset, request, view=None):  # type: ignore[no-untyped-def]
+    def paginate_queryset(
+        self, dynamo_queryset: Any, request: Any, view: Optional[Any] = None
+    ) -> Optional[List[Any]]:
         last_evaluated_key = dynamo_queryset.get("LastEvaluatedKey")
         if last_evaluated_key:
             self.last_evaluated_key = base64.b64encode(
@@ -81,7 +85,7 @@ def paginate_queryset(self, dynamo_queryset, request, view=None):  # type: ignor
             for identity_document in dynamo_queryset["Items"]
         ]
 
-    def get_paginated_response(self, data) -> Response:  # type: ignore[no-untyped-def]
+    def get_paginated_response(self, data: Any) -> Response:
         """
         Note: "If the size of the Query result set is larger than 1 MB, ScannedCount
                 and Count represent only a partial count of the total items"

api/app/routers.py

@@ -1,10 +1,12 @@
 import logging
 import random
 from enum import Enum
+from typing import Any, Type
 
 from django.conf import settings
 from django.core.cache import cache
 from django.db import connections
+from django.db.models import Model
 
 from .exceptions import ImproperlyConfiguredError
 
@@ -47,7 +49,7 @@ def connection_check(database: str) -> bool:
 
 
 class PrimaryReplicaRouter:
-    def db_for_read(self, model, **hints):  # type: ignore[no-untyped-def]
+    def db_for_read(self, model: Type[Model], **hints: Any) -> str | None:
         if settings.NUM_DB_REPLICAS == 0:
             return "default"
 
@@ -75,10 +77,12 @@ def db_for_read(self, model, **hints):  # type: ignore[no-untyped-def]
         )
         return "default"
 
-    def db_for_write(self, model, **hints):  # type: ignore[no-untyped-def]
+    def db_for_write(self, model: Type[Model], **hints: Any) -> str | None:
         return "default"
 
-    def allow_relation(self, obj1, obj2, **hints):  # type: ignore[no-untyped-def]
+    def allow_relation(
+        self, obj1: Type[Model], obj2: Type[Model], **hints: Any
+    ) -> bool | None:
         """
         Relations between objects are allowed if both objects are
         in the primary/replica pool.
@@ -95,10 +99,12 @@ def allow_relation(self, obj1, obj2, **hints):  # type: ignore[no-untyped-def]
             return True
         return None
 
-    def allow_migrate(self, db, app_label, model_name=None, **hints):  # type: ignore[no-untyped-def]
+    def allow_migrate(
+        self, db: str, app_label: str, model_name: str | None = None, **hints: Any
+    ) -> bool | None:
         return db == "default"
 
-    def _get_replica(self, replicas: list[str]) -> None | str:  # type: ignore[return]
+    def _get_replica(self, replicas: list[str]) -> None | str:
         while replicas:
             if settings.REPLICA_READ_STRATEGY == ReplicaReadStrategy.DISTRIBUTED:
                 database = random.choice(replicas)
@@ -119,27 +125,32 @@ def _get_replica(self, replicas: list[str]) -> None | str:  # type: ignore[retur
             if connection_check(database):
                 return database
 
+        # If no replicas are available, return None
+        return None
+
 
 class AnalyticsRouter:
     route_app_labels = ["app_analytics"]
 
-    def db_for_read(self, model, **hints):  # type: ignore[no-untyped-def]
+    def db_for_read(self, model: Type[Model], **hints: Any) -> str | None:
         """
         Attempts to read analytics models go to 'analytics' database.
         """
         if model._meta.app_label in self.route_app_labels:
             return "analytics"
         return None
 
-    def db_for_write(self, model, **hints):  # type: ignore[no-untyped-def]
+    def db_for_write(self, model: Type[Model], **hints: Any) -> str | None:
         """
         Attempts to write analytics models go to 'analytics' database.
         """
         if model._meta.app_label in self.route_app_labels:
             return "analytics"
         return None
 
-    def allow_relation(self, obj1, obj2, **hints):  # type: ignore[no-untyped-def]
+    def allow_relation(
+        self, obj1: Type[Model], obj2: Type[Model], **hints: Any
+    ) -> bool | None:
         """
         Relations between objects are allowed if both objects are
         in the analytics database.
@@ -151,7 +162,9 @@ def allow_relation(self, obj1, obj2, **hints):  # type: ignore[no-untyped-def]
             return True
         return None
 
-    def allow_migrate(self, db, app_label, model_name=None, **hints):  # type: ignore[no-untyped-def]
+    def allow_migrate(
+        self, db: str, app_label: str, model_name: str | None = None, **hints: Any
+    ) -> bool | None:
         """
         Make sure the analytics app only appears in the 'analytics' database
         """

api/tests/unit/app/test_unit_app_routers.py

@@ -39,12 +39,12 @@ def test_replica_router_db_for_read_with_one_offline_replica(
     router = PrimaryReplicaRouter()
 
     # When
-    result = router.db_for_read(FFAdminUser)  # type: ignore[no-untyped-call]
+    result = router.db_for_read(FFAdminUser)
 
     # Then
     # Read strategy DISTRIBUTED is random, so just this is a check
     # against loading the primary or one of the cross region replicas
-    assert result.startswith("replica_")
+    assert result is not None and result.startswith("replica_")
 
     # Check that the number of replica call counts is as expected.
     conn_call_count = 2
@@ -85,12 +85,12 @@ def test_replica_router_db_for_read_with_local_offline_replicas(
     router = PrimaryReplicaRouter()
 
     # When
-    result = router.db_for_read(FFAdminUser)  # type: ignore[no-untyped-call]
+    result = router.db_for_read(FFAdminUser)
 
     # Then
     # Read strategy DISTRIBUTED is random, so just this is a check
     # against loading the primary or one of the cross region replicas
-    assert result.startswith("cross_region_replica_")
+    assert result is not None and result.startswith("cross_region_replica_")
 
     # Check that the number of replica call counts is as expected.
     conn_call_count = 6
@@ -120,7 +120,7 @@ def test_replica_router_db_for_read_with_all_offline_replicas(
     router = PrimaryReplicaRouter()
 
     # When
-    result = router.db_for_read(FFAdminUser)  # type: ignore[no-untyped-call]
+    result = router.db_for_read(FFAdminUser)
 
     # Then
     # Fallback to primary database if all replicas are offline.
@@ -154,7 +154,7 @@ def test_replica_router_db_with_sequential_read(
     router = PrimaryReplicaRouter()
 
     # When
-    result = router.db_for_read(FFAdminUser)  # type: ignore[no-untyped-call]
+    result = router.db_for_read(FFAdminUser)
 
     # Then
     # Fallback from first replica to second one.
@@ -186,7 +186,7 @@ def test_replica_router_db_no_replicas(
     router = PrimaryReplicaRouter()
 
     # When
-    result = router.db_for_read(FFAdminUser)  # type: ignore[no-untyped-call]
+    result = router.db_for_read(FFAdminUser)
 
     # Then
     # Should always use primary database.