Use a typing.Protocol for moderated content models

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: aapeliv

app/backend/src/couchers/models/moderation.py

@@ -9,25 +9,16 @@
 from dataclasses import dataclass
 from datetime import datetime
 from functools import cache
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Protocol
 
 from sqlalchemy import BigInteger, DateTime, Enum, ForeignKey, Index, String, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from couchers.models.base import Base, moderation_seq
 
 if TYPE_CHECKING:
-    from couchers.models.conversations import GroupChat
-    from couchers.models.discussions import Comment, Discussion, Reply
-    from couchers.models.events import EventOccurrence
-    from couchers.models.host_requests import HostRequest
-    from couchers.models.rest import FriendRelationship
     from couchers.models.users import User
 
-    type ModeratedContentModel = type[
-        HostRequest | GroupChat | FriendRelationship | EventOccurrence | Comment | Reply | Discussion
-    ]
-
 
 class ModerationVisibility(enum.Enum):
     # Only visible to moderators
@@ -199,14 +190,23 @@ def __repr__(self) -> str:
         return f"ModerationLog(id={self.id}, state_id={self.moderation_state_id}, action={self.action}, moderator={self.moderator_user_id}, time={self.time})"
 
 
+class ModeratedContent(Protocol):
+    """A model governed by the UMS, identified by the moderation metadata it declares as class attributes."""
+
+    __moderation_object_type__: ModerationObjectType
+    __moderation_author_column__: str
+
+
 @dataclass(frozen=True)
 class ModeratedModel:
     """A model governed by the UMS, with its moderation metadata resolved."""
 
     object_type: ModerationObjectType
-    model: ModeratedContentModel
-    # the InstrumentedAttribute of the model's author column; Any avoids descriptor-unwrapping on access
+    model: type[ModeratedContent]
+    # InstrumentedAttributes/Columns resolved from the model; Any avoids descriptor-unwrapping on access
     author_column: Any
+    object_id_column: Any
+    moderation_state_id_column: Any
 
 
 @cache
@@ -222,9 +222,12 @@ def get_moderated_models() -> dict[ModerationObjectType, ModeratedModel]:
         cls = mapper.class_
         if not hasattr(cls, "__moderation_object_type__"):
             continue
-        models[cls.__moderation_object_type__] = ModeratedModel(
-            object_type=cls.__moderation_object_type__,
-            model=cls,
-            author_column=getattr(cls, cls.__moderation_author_column__),
+        model: type[ModeratedContent] = cls
+        models[model.__moderation_object_type__] = ModeratedModel(
+            object_type=model.__moderation_object_type__,
+            model=model,
+            author_column=getattr(model, model.__moderation_author_column__),
+            object_id_column=mapper.primary_key[0],
+            moderation_state_id_column=mapper.columns["moderation_state_id"],
         )
     return models

app/backend/src/couchers/servicers/moderation.py

@@ -135,7 +135,7 @@ def bulk_set_user_content_visibility(
     author_exists_clauses = []
     for entry in get_moderated_models().values():
         author_exists_clauses.append(
-            exists().where(and_(entry.model.moderation_state_id == ModerationState.id, entry.author_column == user.id))
+            exists().where(and_(entry.moderation_state_id_column == ModerationState.id, entry.author_column == user.id))
         )
 
     states = session.execute(select(ModerationState).where(or_(*author_exists_clauses))).scalars().all()
@@ -335,7 +335,7 @@ def GetModerationQueue(
                 author_exists_clauses.append(
                     exists().where(
                         and_(
-                            entry.model.moderation_state_id == ModerationQueueItem.moderation_state_id,
+                            entry.moderation_state_id_column == ModerationQueueItem.moderation_state_id,
                             entry.author_column == author_user_id,
                         )
                     )
@@ -675,7 +675,7 @@ def ListModerationStates(
                 author_exists_clauses.append(
                     exists().where(
                         and_(
-                            entry.model.moderation_state_id == ModerationState.id,
+                            entry.moderation_state_id_column == ModerationState.id,
                             entry.author_column == request.author_user_id,
                         )
                     )

app/backend/src/couchers/sql.py

@@ -17,7 +17,7 @@
 
 if TYPE_CHECKING:
     from couchers.materialized_views import LiteUser
-    from couchers.models.moderation import ModeratedContentModel
+    from couchers.models.moderation import ModeratedContent
 
     type _UserLike = type[User | LiteUser | SignupFlow]
     type _User = type[User | LiteUser]
@@ -139,10 +139,11 @@ def where_user_columns_visible_to_each_other[T: tuple[Any, ...]](
 
 def where_moderated_content_visible_to_user_column[T: tuple[Any, ...]](
     query: Select[T],
-    table: ModeratedContentModel,
+    table: type[ModeratedContent],
     user_id_column: InstrumentedAttribute[int],
     is_list_operation: bool = False,
 ) -> Select[T]:
+    entry = get_moderated_models()[table.__moderation_object_type__]
     aliased_mod_state = aliased(ModerationState)
     conditions = [aliased_mod_state.visibility == ModerationVisibility.visible]
 
@@ -154,19 +155,22 @@ def where_moderated_content_visible_to_user_column[T: tuple[Any, ...]](
     conditions.append(
         and_(
             aliased_mod_state.visibility == ModerationVisibility.shadowed,
-            getattr(table, table.__moderation_author_column__) == user_id_column,
+            entry.author_column == user_id_column,
         )
     )
 
-    return query.join(aliased_mod_state, aliased_mod_state.id == table.moderation_state_id).where(or_(*conditions))
+    return query.join(aliased_mod_state, aliased_mod_state.id == entry.moderation_state_id_column).where(
+        or_(*conditions)
+    )
 
 
 def where_moderated_content_visible[T: tuple[Any, ...]](
     query: Select[T],
     context: CouchersContext,
-    table: ModeratedContentModel,
+    table: type[ModeratedContent],
     is_list_operation: bool = False,
 ) -> Select[T]:
+    entry = get_moderated_models()[table.__moderation_object_type__]
     aliased_mod_state = aliased(ModerationState)
     conditions = [aliased_mod_state.visibility == ModerationVisibility.visible]
 
@@ -179,11 +183,13 @@ def where_moderated_content_visible[T: tuple[Any, ...]](
         conditions.append(
             and_(
                 aliased_mod_state.visibility == ModerationVisibility.shadowed,
-                getattr(table, table.__moderation_author_column__) == context.user_id,
+                entry.author_column == context.user_id,
             )
         )
 
-    return query.join(aliased_mod_state, aliased_mod_state.id == table.moderation_state_id).where(or_(*conditions))
+    return query.join(aliased_mod_state, aliased_mod_state.id == entry.moderation_state_id_column).where(
+        or_(*conditions)
+    )
 
 
 def moderation_state_column_visible(
@@ -207,14 +213,13 @@ def moderation_state_column_visible(
     shadowed_conditions: list[ColumnElement[bool]] = []
     if context.is_logged_in():
         for entry in get_moderated_models().values():
-            object_id_column = entry.model.__mapper__.primary_key[0]
             shadowed_conditions.append(
                 and_(
                     aliased_mod_state.object_type == entry.object_type,
                     exists(
                         select(1)
                         .select_from(entry.model)
-                        .where(object_id_column == aliased_mod_state.object_id)
+                        .where(entry.object_id_column == aliased_mod_state.object_id)
                         .where(entry.author_column == context.user_id)
                     ),
                 )