feat: soft-deletion of database objects (#102)

* feat: database softdelete
* feat: cascade delete
* feat: add cascading delete on BiocommonsGroup and Platform
* feat: add migration script
* fix: add warning to migration

Author: uwwint

db/core.py

@@ -1,7 +1,14 @@
-from typing import ClassVar
+from __future__ import annotations
 
-from sqlalchemy import MetaData
-from sqlmodel import SQLModel
+from typing import Any, ClassVar
+
+from sqlalchemy import MetaData, event, select
+from sqlalchemy import inspect as sa_inspect
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session as SASession
+from sqlalchemy.orm import with_loader_criteria
+from sqlalchemy.sql import expression
+from sqlmodel import Field, Session, SQLModel
 
 naming_convention = {
     "ix": "ix_%(column_0_label)s",
@@ -27,4 +34,154 @@ class BaseModel(SQLModel):
     metadata: ClassVar[MetaData] = metadata
 
 
-__all__ = ["BaseModel"]
+class SoftDeleteModel(BaseModel):
+    """
+    Base for ORM models that should support soft deletion.
+    """
+    __abstract__ = True
+
+    is_deleted: bool = Field(
+        default=False,
+        nullable=False,
+        index=True,
+        sa_column_kwargs={"server_default": expression.false()},
+        description="Soft-delete flag. True means the row is hidden from default queries.",
+    )
+
+    def delete(self, session: Session, commit: bool = False) -> "SoftDeleteModel":
+        """
+        Soft delete this record (mark as deleted without removing from DB).
+        """
+        self.is_deleted = True
+        session.add(self)
+        if commit:
+            session.commit()
+            session.expunge(self)
+        return self
+
+    def restore(self, session: Session, commit: bool = False) -> "SoftDeleteModel":
+        """
+        Restore a previously deleted record.
+        """
+        self.is_deleted = False
+        session.add(self)
+        if commit:
+            session.commit()
+        return self
+
+    @classmethod
+    def get_deleted_by_id(cls, session: Session, identity: Any) -> "SoftDeleteModel | None":
+        """
+        Retrieve a soft-deleted record by primary key.
+        """
+        identity_dict = cls._coerce_primary_key_map(identity)
+        stmt = (
+            select(cls)
+            .execution_options(include_deleted=True)
+            .filter_by(**identity_dict)
+            .where(cls.is_deleted.is_(True))
+        )
+        return session.exec(stmt).scalars().one_or_none()
+
+    @classmethod
+    def _coerce_primary_key_map(cls, identity: Any) -> dict[str, Any]:
+        """
+        Coerce arbitrary primary key identifiers into a ``{column: value}`` mapping.
+
+        ``identity`` may be provided as:
+        * a dict where keys match the primary key columns,
+        * a single scalar value when the model uses a single-column primary key,
+        * a tuple/list containing values for each primary key column in order.
+
+        Any mismatch between the provided structure and the model's primary key
+        definition raises ``ValueError`` so downstream queries remain predictable.
+        """
+        mapper = sa_inspect(cls)
+        pk_cols = mapper.primary_key
+        if not pk_cols:
+            raise ValueError(f"{cls.__name__} does not have a primary key defined.")
+
+        if isinstance(identity, dict):
+            return identity
+
+        if len(pk_cols) == 1 and not isinstance(identity, (tuple, list)):
+            return {pk_cols[0].key: identity}
+
+        if isinstance(identity, (tuple, list)):
+            if len(identity) != len(pk_cols):
+                raise ValueError(
+                    f"Identity length {len(identity)} does not match primary key length {len(pk_cols)}"
+                )
+            return {col.key: value for col, value in zip(pk_cols, identity)}
+
+        raise ValueError("Identity must be scalar, tuple/list, or dict matching the primary key.")
+
+
+def _copy_column_state(source: SoftDeleteModel, target: SoftDeleteModel) -> None:
+    mapper = sa_inspect(source.__class__)
+    for attr in mapper.column_attrs:
+        key = attr.key
+        if key == "is_deleted":
+            continue
+        setattr(target, key, getattr(source, key))
+
+
+def _identity_dict_from_instance(instance: SoftDeleteModel) -> dict[str, Any] | None:
+    mapper = sa_inspect(instance.__class__)
+    identity: dict[str, Any] = {}
+    for column in mapper.primary_key:
+        value = getattr(instance, column.key, None)
+        if value is None:
+            return None
+        identity[column.key] = value
+    return identity
+
+
+def _soft_delete_filter(cls) -> Any:
+    mapper = sa_inspect(cls, raiseerr=False)
+    if mapper is None:
+        return expression.true()
+    column = mapper.c.is_deleted
+    return column.is_(False)
+
+
+@event.listens_for(SASession, "before_flush")
+def _revive_soft_deleted(session: SASession, flush_context, instances) -> None:
+    for obj in list(session.new):
+        if not isinstance(obj, SoftDeleteModel):
+            continue
+        identity_dict = _identity_dict_from_instance(obj)
+        if identity_dict is None:
+            continue
+        stmt = (
+            select(obj.__class__)
+            .execution_options(include_deleted=True)
+            .filter_by(**identity_dict)
+        )
+        existing = session.exec(stmt).scalars().one_or_none()
+        if existing is None:
+            continue
+        if not existing.is_deleted:
+            raise IntegrityError(
+                "Duplicate primary key for active record",
+                params=None,
+                orig=None,
+            )
+        _copy_column_state(obj, existing)
+        existing.is_deleted = False
+        session.expunge(obj)
+
+
+@event.listens_for(SASession, "do_orm_execute")
+def _filter_soft_deleted(execute_state) -> None:
+    if execute_state.is_select and not execute_state.execution_options.get("include_deleted", False):
+        execute_state.statement = execute_state.statement.options(
+            with_loader_criteria(
+                SoftDeleteModel,
+                lambda cls: _soft_delete_filter(cls),
+                include_aliases=True,
+            )
+        )
+
+
+__all__ = ["BaseModel", "SoftDeleteModel"]

db/models.py

@@ -9,7 +9,7 @@
 
 import schemas
 from auth0.client import Auth0Client
-from db.core import BaseModel
+from db.core import SoftDeleteModel
 from db.types import (
     ApprovalStatusEnum,
     GroupMembershipData,
@@ -19,7 +19,7 @@
 from schemas.user import SessionUser
 
 
-class BiocommonsUser(BaseModel, table=True):
+class BiocommonsUser(SoftDeleteModel, table=True):
     __tablename__ = "biocommons_user"
     # Auth0 ID
     id: str = Field(primary_key=True)
@@ -50,13 +50,14 @@ def has_platform_membership(cls, user_id: str, platform_id: PlatformEnum, sessio
         """
         Check if a user has a membership for a specific platform.
         """
-        return session.exec(
-            select(PlatformMembership).where(
+        result = session.exec(
+            select(PlatformMembership.id).where(
                 PlatformMembership.user_id == user_id,
                 PlatformMembership.platform_id == platform_id,
                 PlatformMembership.approval_status == ApprovalStatusEnum.APPROVED,
             )
-        ).exists()
+        ).first()
+        return result is not None
 
     @classmethod
     def create_from_auth0(cls, auth0_id: str, auth0_client: Auth0Client) -> Self:
@@ -87,6 +88,20 @@ def get_or_create(
             db_session.commit()
         return user
 
+    def delete(self, session: Session, commit: bool = False) -> "BiocommonsUser":
+        """
+        Soft delete the user and cascade the soft delete to related memberships.
+        """
+        for membership in list(self.platform_memberships or []):
+            if not membership.is_deleted:
+                membership.delete(session, commit=False)
+        for membership in list(self.group_memberships or []):
+            if not membership.is_deleted:
+                membership.delete(session, commit=False)
+
+        super().delete(session, commit=commit)
+        return self
+
     def update_from_auth0(self, auth0_id: str, auth0_client: Auth0Client) -> Self:
         """
         Fetch user data from Auth0 and update this object with it.
@@ -133,12 +148,12 @@ def add_group_membership(
         return membership
 
 
-class PlatformRoleLink(BaseModel, table=True):
+class PlatformRoleLink(SoftDeleteModel, table=True):
     platform_id: PlatformEnum = Field(primary_key=True, foreign_key="platform.id", sa_type=DbEnum(PlatformEnum, name="PlatformEnum"))
     role_id: str = Field(primary_key=True, foreign_key="auth0role.id")
 
 
-class Platform(BaseModel, table=True):
+class Platform(SoftDeleteModel, table=True):
     id: PlatformEnum = Field(primary_key=True, unique=True, sa_type=DbEnum(PlatformEnum, name="PlatformEnum"))
     # Human-readable name for the platform
     name: str = Field(unique=True)
@@ -168,8 +183,17 @@ def get_approved_by_user_id(cls, user_id: str, session: Session) -> list[Self] |
             .where(PlatformMembership.approval_status == ApprovalStatusEnum.APPROVED)
         ).all()
 
+    def delete(self, session: Session, commit: bool = False) -> "Platform":
+        memberships = list(self.members or [])
+        for membership in memberships:
+            if not membership.is_deleted:
+                membership.delete(session, commit=False)
+
+        super().delete(session, commit=commit)
+        return self
+
 
-class PlatformMembership(BaseModel, table=True):
+class PlatformMembership(SoftDeleteModel, table=True):
     __table_args__ = (
         UniqueConstraint("platform_id", "user_id", name="platform_user_id_platform_id"),
     )
@@ -224,6 +248,26 @@ def get_by_user_id_and_platform_id(cls, user_id: str, platform_id: PlatformEnum,
             )
         ).one_or_none()
 
+    def delete(self, session: Session, commit: bool = False) -> "PlatformMembership":
+        history_entries = session.exec(
+            select(PlatformMembershipHistory)
+            .where(
+                PlatformMembershipHistory.user_id == self.user_id,
+                PlatformMembershipHistory.platform_id == self.platform_id,
+            )
+        ).all()
+
+        super().delete(session, commit=False)
+
+        for history in history_entries:
+            if not history.is_deleted:
+                history.delete(session, commit=False)
+
+        if commit:
+            session.commit()
+            session.expunge(self)
+        return self
+
     def save_history(self, session: Session) -> "PlatformMembershipHistory":
         # Make sure this object is in the session before accessing relationships
         if self not in session:
@@ -261,7 +305,7 @@ def get_data(self) -> PlatformMembershipData:
 
 
 
-class PlatformMembershipHistory(BaseModel, table=True):
+class PlatformMembershipHistory(SoftDeleteModel, table=True):
     id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
     platform_id: PlatformEnum = Field(sa_type=DbEnum(PlatformEnum, name="PlatformEnum"))
     user_id: str = Field(foreign_key="biocommons_user.id")
@@ -288,7 +332,7 @@ class PlatformMembershipHistory(BaseModel, table=True):
     )
 
 
-class GroupMembership(BaseModel, table=True):
+class GroupMembership(SoftDeleteModel, table=True):
     """
     Stores the current approval status for a user/group pairing.
     Note: only one row per user/group, the approval history
@@ -351,15 +395,36 @@ def get_by_user_id_and_group_id(cls, user_id: str, group_id: str, session: Sessi
             )
         ).one_or_none()
 
+    def delete(self, session: Session, commit: bool = False) -> "GroupMembership":
+        history_entries = session.exec(
+            select(GroupMembershipHistory)
+            .where(
+                GroupMembershipHistory.user_id == self.user_id,
+                GroupMembershipHistory.group_id == self.group_id,
+            )
+        ).all()
+
+        super().delete(session, commit=False)
+
+        for history in history_entries:
+            if not history.is_deleted:
+                history.delete(session, commit=False)
+
+        if commit:
+            session.commit()
+            session.expunge(self)
+        return self
+
     @classmethod
     def has_group_membership(cls, user_id: str, group_id: str, session: Session) -> bool:
-        return session.exec(
-            select(GroupMembership).where(
+        result = session.exec(
+            select(GroupMembership.id).where(
                 GroupMembership.user_id == user_id,
                 GroupMembership.group_id == group_id,
                 GroupMembership.approval_status == ApprovalStatusEnum.APPROVED,
             )
-        ).exists()
+        ).first()
+        return result is not None
 
 
 
@@ -422,7 +487,7 @@ def get_data(self) -> GroupMembershipData:
 
 
 
-class GroupMembershipHistory(BaseModel, table=True):
+class GroupMembershipHistory(SoftDeleteModel, table=True):
     """
     Stores the full history of approval decisions for each user
     """
@@ -478,12 +543,12 @@ def get_by_user_id(cls, user_id: str, session: Session) -> list[Self] | None:
         ).all()
 
 
-class GroupRoleLink(BaseModel, table=True):
+class GroupRoleLink(SoftDeleteModel, table=True):
     group_id: str = Field(primary_key=True, foreign_key="biocommonsgroup.group_id")
     role_id: str = Field(primary_key=True, foreign_key="auth0role.id")
 
 
-class Auth0Role(BaseModel, table=True):
+class Auth0Role(SoftDeleteModel, table=True):
     id: str = Field(primary_key=True, unique=True)
     name: str
     description: str = Field(default="")
@@ -529,7 +594,7 @@ def get_or_create_by_name(
         return role
 
 
-class BiocommonsGroup(BaseModel, table=True):
+class BiocommonsGroup(SoftDeleteModel, table=True):
     # Name of the group / role name in Auth0, e.g. biocommons/group/tsi
     group_id: str = Field(primary_key=True, unique=True)
     # Human-readable name for the group
@@ -549,6 +614,15 @@ class BiocommonsGroup(BaseModel, table=True):
     def get_by_id(cls, group_id: str, session: Session) -> Self | None:
         return session.get(BiocommonsGroup, group_id)
 
+    def delete(self, session: Session, commit: bool = False) -> "BiocommonsGroup":
+        memberships = list(self.members or [])
+        for membership in memberships:
+            if not membership.is_deleted:
+                membership.delete(session, commit=False)
+
+        super().delete(session, commit=commit)
+        return self
+
     def get_admins(self, auth0_client: Auth0Client) -> set[str]:
         """
         Get all admin emails for this group from the Auth0 API, returning a set of emails.