Merge remote-tracking branch 'origin/main' into claude/resolve-issue-1332-1lh9h

# Conflicts:
#	config/graphql/analysis_mutations.py
#	config/graphql/annotation_types.py
#	config/graphql/base_types.py
#	config/graphql/corpus_queries.py
#	config/graphql/document_queries.py
#	config/graphql/filters.py
#	config/graphql/moderation_mutations.py
#	config/graphql/og_metadata_queries.py
#	config/graphql/pipeline_queries.py
#	config/graphql/security.py
#	config/graphql/slug_queries.py
#	config/graphql/user_queries.py
#	config/graphql/user_types.py
#	docs/typing/mypy_baseline.txt
#	mypy.ini

Author: JSv4

CHANGELOG.md

@@ -2,6 +2,8 @@
 Serializers related to annotations to avoid circular imports.
 """
 
+from typing import Any
+
 from django.contrib.auth import get_user_model
 from rest_framework import serializers
 
@@ -31,7 +33,7 @@ class Meta:
         ]
         read_only_fields = ["id", "creator"]
 
-    def create(self, validated_data):
+    def create(self, validated_data) -> Any:
         creator_id = validated_data.pop("creator_id", None)
         if creator_id:
             try:

config/graphql/annotation_serializers.py

@@ -2,6 +2,8 @@
 import logging
 import traceback
 from abc import ABC
+from collections.abc import Sequence
+from typing import Any, ClassVar, Optional
 
 import django.db.models
 import graphene
@@ -21,12 +23,30 @@
 logger = logging.getLogger(__name__)
 
 
+def _require_io_setting(mutation_cls: type, name: str) -> Any:
+    """Raise ``NotImplementedError`` if ``cls.IOSettings.<name>`` is missing or ``None``."""
+    io_settings = getattr(mutation_cls, "IOSettings", None)
+    if io_settings is None:
+        raise NotImplementedError(
+            f"{mutation_cls.__name__} must define an IOSettings class."
+        )
+    value = getattr(io_settings, name, None)
+    if value is None:
+        raise NotImplementedError(
+            f"{mutation_cls.__name__}.IOSettings.{name} must be set by the "
+            f"subclass."
+        )
+    return value
+
+
 class OpenContractsNode(Node):
     class Meta:
         name = "Node"
 
     @classmethod
-    def get_node_from_global_id(cls, info, global_id, only_type=None):
+    def get_node_from_global_id(
+        cls, info: Any, global_id: str, only_type: Any | None = None
+    ) -> Any:
 
         _type, _id = from_global_id(global_id)
 
@@ -63,7 +83,7 @@ class Meta:
 
     total_count = graphene.Int()
 
-    def resolve_total_count(root, info, **kwargs):
+    def resolve_total_count(root, info, **kwargs) -> Any:
         if isinstance(root.iterable, django.db.models.QuerySet):
             return root.iterable.count()
         else:
@@ -72,7 +92,8 @@ def resolve_total_count(root, info, **kwargs):
 
 class DRFDeletion(graphene.Mutation):
     class IOSettings(ABC):
-        lookup_field = "id"
+        lookup_field: ClassVar[str] = "id"
+        model: ClassVar[Optional[type[django.db.models.Model]]] = None
 
     class Arguments:
         id = graphene.String(required=False)
@@ -83,14 +104,26 @@ class Arguments:
     @classmethod
     @login_required
     @graphql_ratelimit(rate=RateLimits.WRITE_LIGHT)
-    def mutate(cls, root, info, *args, **kwargs):
+    def mutate(cls, root, info, *args, **kwargs) -> "DRFDeletion":
 
+        # Unlike ``DRFMutation.mutate`` below, this method intentionally does
+        # NOT wrap the body in ``except Exception``. Errors (including the
+        # ``NotImplementedError`` from ``_require_io_setting`` and the
+        # ``ValueError`` for missing lookup args) propagate to the GraphQL
+        # framework as raw execution errors.
         ok = False
 
-        id = from_global_id(kwargs.get(cls.IOSettings.lookup_field, None))[1]
+        model = _require_io_setting(cls, "model")
+        lookup_field = cls.IOSettings.lookup_field
+        lookup_value = kwargs.get(lookup_field)
+        if lookup_value is None:
+            raise ValueError(
+                f"'{lookup_field}' is required to identify the object to delete."
+            )
+        id = from_global_id(lookup_value)[1]
         # Filter through visible_to_user() to prevent IDOR -- returns same
         # DoesNotExist error whether object is missing or user lacks access.
-        obj = cls.IOSettings.model.objects.visible_to_user(info.context.user).get(pk=id)
+        obj = model.objects.visible_to_user(info.context.user).get(pk=id)
 
         # if there's a user lock, only the lock holder (or superuser) can proceed
         if hasattr(obj, "user_lock") and obj.user_lock is not None:
@@ -123,11 +156,13 @@ def mutate(cls, root, info, *args, **kwargs):
 
 class DRFMutation(graphene.Mutation):
     class IOSettings(ABC):
-        pk_fields: list[str | int] = []
-        lookup_field = "id"
-        model: django.db.models.Model = None
-        graphene_model: DjangoObjectType = None
-        serializer = None
+        # Frozen default — subclasses override with their own list/tuple.
+        # Using a tuple here avoids the shared-mutable-default footgun.
+        pk_fields: ClassVar[Sequence[str]] = ()
+        lookup_field: ClassVar[str] = "id"
+        model: ClassVar[Optional[type[django.db.models.Model]]] = None
+        graphene_model: ClassVar[Optional[type[DjangoObjectType]]] = None
+        serializer: ClassVar[Optional[type[serializers.Serializer]]] = None
 
     class Arguments:
         pass
@@ -137,7 +172,7 @@ class Arguments:
     obj_id = graphene.ID()
 
     @staticmethod
-    def format_validation_error(ve):
+    def format_validation_error(ve: serializers.ValidationError) -> str:
         """Surface validation errors with clean formatting.
 
         ``str(ValidationError)`` renders as
@@ -158,7 +193,7 @@ def format_validation_error(ve):
     @classmethod
     @login_required
     @graphql_ratelimit(rate=RateLimits.WRITE_MEDIUM)
-    def mutate(cls, root, info, *args, **kwargs):
+    def mutate(cls, root, info, *args, **kwargs) -> "DRFMutation":
 
         ok = False
         obj_id = None
@@ -175,20 +210,25 @@ def mutate(cls, root, info, *args, **kwargs):
                 raise ValueError("No user in this request...")
 
             logger.info(f"DRFMutation - kwargs: {kwargs}")
-            serializer = cls.IOSettings.serializer
+            serializer = _require_io_setting(cls, "serializer")
+            model = _require_io_setting(cls, "model")
+            # ``graphene_model`` is the class itself; ``__name__`` is the GraphQL
+            # type name (e.g. ``"CorpusType"``). ``__class__.__name__`` would
+            # return the metaclass name (``"SubclassWithMeta_Meta"``) which is
+            # the bug this PR fixes — kept dereferenced as ``__name__`` below.
+            graphene_model = _require_io_setting(cls, "graphene_model")
 
             if hasattr(cls.IOSettings, "pk_fields"):
                 for pk_field in cls.IOSettings.pk_fields:
                     if pk_field in kwargs:
-                        if isinstance(kwargs[pk_field], list):
-                            pk_value = []
-                            for global_id in kwargs[pk_field]:
-                                # global_id is already the ID string, not a key
-                                pk_value.append(from_global_id(global_id)[1])
+                        raw_value = kwargs[pk_field]
+                        if isinstance(raw_value, list):
+                            kwargs[pk_field] = [
+                                from_global_id(global_id)[1] for global_id in raw_value
+                            ]
                         else:
-                            logger.info(f"pk field is: {kwargs.get(pk_field, None)}")
-                            pk_value = from_global_id(kwargs.get(pk_field, None))[1]
-                        kwargs[pk_field] = pk_value
+                            logger.info(f"pk field is: {raw_value}")
+                            kwargs[pk_field] = from_global_id(raw_value)[1]
 
             # Check if lookup_field exists in IOSettings and if it's in kwargs
             # This allows create mutations to work without requiring lookup_field
@@ -201,10 +241,8 @@ def mutate(cls, root, info, *args, **kwargs):
                 logger.info("Lookup_field specified - update")
                 # Filter through visible_to_user() to prevent IDOR --
                 # returns same DoesNotExist whether missing or no access.
-                obj = cls.IOSettings.model.objects.visible_to_user(
-                    info.context.user
-                ).get(
-                    pk=from_global_id(kwargs.get(cls.IOSettings.lookup_field, None))[1]
+                obj = model.objects.visible_to_user(info.context.user).get(
+                    pk=from_global_id(kwargs[cls.IOSettings.lookup_field])[1]
                 )
 
                 logger.info(f"Retrieved obj: {obj}")
@@ -240,9 +278,7 @@ def mutate(cls, root, info, *args, **kwargs):
                 obj_serializer.save()
                 ok = True
                 message = "Success"
-                obj_id = to_global_id(
-                    cls.IOSettings.graphene_model.__class__.__name__, obj.id
-                )
+                obj_id = to_global_id(graphene_model.__name__, obj.id)
                 logger.info("Succeeded updating obj")
 
             else:
@@ -262,9 +298,7 @@ def mutate(cls, root, info, *args, **kwargs):
 
                 ok = True
                 message = "Success"
-                obj_id = to_global_id(
-                    cls.IOSettings.graphene_model.__class__.__name__, obj.id
-                )
+                obj_id = to_global_id(graphene_model.__name__, obj.id)
 
         except serializers.ValidationError as ve:
             logger.warning(f"Validation error in mutation: {ve.detail}")

config/graphql/base.py

@@ -142,7 +142,7 @@ def resolve_mentioned_resources(self, info) -> Any:
         from opencontractserver.corpuses.models import Corpus
         from opencontractserver.documents.models import Document, DocumentPath
 
-        def _extract_annotation_id(url: str):
+        def _extract_annotation_id(url: str) -> Any:
             """
             Extract annotation ID from URL query params.
 

config/graphql/conversation_types.py

@@ -3,6 +3,7 @@
 """
 
 import logging
+from typing import Any
 
 import graphene
 from django.conf import settings
@@ -536,7 +537,7 @@ def dispatch_fork_task(
                 _corpus_id=corpus.id,
                 _collected=collected,
                 _user_id=info.context.user.id,
-            ):
+            ) -> Any:
                 fork_corpus.si(
                     _corpus_id,
                     _collected.document_ids,

config/graphql/corpus_mutations.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any
 
 from graphene import Connection, Int
 
@@ -12,14 +13,14 @@ class Meta:
     current_page = Int()
     page_count = Int()
 
-    def resolve_current_page(root, info, **kwargs):
+    def resolve_current_page(root, info, **kwargs) -> Any:
         # print(
         #     f"PdfPageAwareConnection- resolve_total_count kwargs: {kwargs} / root {dir(root)} / iteracble "
         #     f"{root.iterable.count()}"
         # )
         return 1
 
-    def resolve_page_count(root, info, **kwargs):
+    def resolve_page_count(root, info, **kwargs) -> Any:
 
         largest_page_number = max(
             list(root.iterable.values_list("page", flat=True).distinct())

config/graphql/custom_connections.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
+from typing import Any
 
 from graphql_relay import from_global_id
 
@@ -46,7 +47,7 @@ def _apply_filter(sequence: Iterable, predicate) -> list:
     return [item for item in sequence if predicate(item)]
 
 
-def resolve_doc_annotations_optimized(self, info, **kwargs):
+def resolve_doc_annotations_optimized(self, info, **kwargs) -> Any:
     """Resolve ``docAnnotations`` while favouring prefetched data and the optimizer."""
 
     if kwargs.get("after") or kwargs.get("before"):

config/graphql/custom_resolvers.py

@@ -123,7 +123,7 @@ class Meta:
     _VISIBLE_CORPUS_IDS_CACHE_KEY = "_docpath_visible_corpus_ids"
 
     @classmethod
-    def _get_visible_corpus_ids(cls, info):
+    def _get_visible_corpus_ids(cls, info) -> Any:
         """Get visible corpus IDs with request-level caching to prevent N+1 queries."""
         from opencontractserver.corpuses.models import Corpus
 
@@ -1179,19 +1179,19 @@ class Meta:
         connection_class = CountableConnection
 
 
-def _get_corpus_folder_type():
+def _get_corpus_folder_type() -> Any:
     from config.graphql.corpus_types import CorpusFolderType
 
     return CorpusFolderType
 
 
-def _get_corpus_action_type():
+def _get_corpus_action_type() -> Any:
     from config.graphql.agent_types import CorpusActionType
 
     return CorpusActionType
 
 
-def _get_extract_type():
+def _get_extract_type() -> Any:
     from config.graphql.extract_types import ExtractType
 
     return ExtractType

config/graphql/document_types.py

@@ -64,7 +64,7 @@ class Meta:
         connection_class = CountableConnection
 
 
-def _get_datacell_qs(extract, user):
+def _get_datacell_qs(extract, user) -> Any:
     """Return the permission-filtered, deterministically ordered queryset.
 
     Note: this is a module-level function because Graphene-Django resolvers

config/graphql/extract_types.py

@@ -3,6 +3,7 @@
 """
 
 import logging
+from typing import Any
 
 import graphene
 from django.db import IntegrityError
@@ -45,7 +46,7 @@ def _parse_ingestion_source_global_id(
     return pk, None
 
 
-def _resolve_source_type(source_type):
+def _resolve_source_type(source_type) -> Any:
     """Coerce a graphene Enum to its string value, defaulting to MANUAL."""
     if source_type is None:
         return IngestionSourceCategory.MANUAL