"""Soft Delete Mixin.""" from typing import Any, Sequence from sqlalchemy import Select, update from ..contexts import get_user_id from ..errors import ERROR_ENTITY_IS_SOFT_DELETED from ..types.executor import Executor from ..utils import current_timestamp, try_pk_column, try_soft_deletable class SoftDeleteMixin: """Mixin for models with soft delete tracking. The actual columns are defined in the model itself based on what exists in the database schema. Soft deletion marks records as deleted without removing them from the database. Query filters should check for deleted_at / deleted_by IS NULL to get active records. A model using this mixin may have one or more of: - deleted_at: When the record was soft-deleted (NULL if active) - deleted_by: User ID who soft-deleted the record (NULL if active) If a model has both deleted_at and deleted_by, deleted_at is used as the source of truth for determining deleted status. """ @classmethod def filter_active(cls, stmt: Select) -> Select: """Filter query to return only active (non-deleted) records. Active records must have their primary deletion field not set (NULL). Usage: ```python stmt = select(MyModel) stmt = MyModel.filter_active(stmt) ``` Raises `TypeError` if model does not support soft deletion """ try_soft_deletable(cls) if hasattr(cls, 'deleted_at'): return stmt.where(cls.deleted_at.is_(None)) # type: ignore[attr-defined] return stmt.where(cls.deleted_by.is_(None)) # type: ignore[attr-defined] @classmethod def filter_deleted(cls, stmt: Select) -> Select: """Filter query to return only deleted records. Deleted records have one or more deletion fields set (not NULL). ```python stmt = select(MyModel) stmt = MyModel.filter_deleted(stmt) ``` """ try_soft_deletable(cls) if hasattr(cls, 'deleted_at'): return stmt.where(cls.deleted_at.is_not(None)) # type: ignore[attr-defined] return stmt.where(cls.deleted_by.is_not(None)) # type: ignore[attr-defined] def is_deleted(self) -> bool: """Check if record is soft-deleted. Returns `True` if `deleted_at` or `deleted_by` is set (not `None`). If neither field exists, raises TypeError. """ try_soft_deletable(self) if hasattr(self, 'deleted_at'): return self.deleted_at is not None return self.deleted_by is not None @classmethod def soft_delete_many(cls, executor: Executor, ids: Sequence[Any]) -> None: """Soft delete multiple records by ID. Uses bulk update for performance. Args: executor: SQLAlchemy Session or Connection ids: List of primary key IDs """ try_soft_deletable(cls) pk_column = try_pk_column(cls) values: dict[str, Any] = {} if hasattr(cls, 'deleted_at'): values['deleted_at'] = current_timestamp() if hasattr(cls, 'deleted_by'): values['deleted_by'] = get_user_id() stmt = update(cls).where(pk_column.in_(ids)).values(**values) executor.execute(stmt) def soft_delete(self) -> None: """Mark this record as deleted. Raises ValueError if the record is already deleted. If the model uses UpdateMixin, last_modified fields will also be updated via SQLAlchemy event handler. """ if self.is_deleted(): raise ValueError(ERROR_ENTITY_IS_SOFT_DELETED) if hasattr(self, 'deleted_at'): self.deleted_at = current_timestamp() if hasattr(self, 'deleted_by'): self.deleted_by = get_user_id() def restore(self) -> None: """Restore a soft-deleted record. If the model uses UpdateMixin, last_modified fields will also be updated via SQLAlchemy event handler. """ try_soft_deletable(self) if hasattr(self, 'deleted_at'): self.deleted_at = None if hasattr(self, 'deleted_by'): self.deleted_by = None