SQLAlchemyMiddleSystem design

Как реализовать soft deletes в SQLAlchemy?

Реализуется через поле deleted_at в модели (Mixin + SoftDeleteMixin). Для автоматической фильтрации используется SQLAlchemy-event do_orm_execute с with_loader_criteria; include_deleted=True позволяет явно включить удалённые записи в конкретном запросе.

Базовая реализация

Добавьте поле deleted_at в модель и фильтруйте через него:

from datetime import datetime
from sqlalchemy import Column, DateTime
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class Post(Base):
    __tablename__ = "posts"

    id         = Column(Integer, primary_key=True)
    title      = Column(String, nullable=False)
    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

    def soft_delete(self) -> None:
        self.deleted_at = datetime.utcnow()

Автоматический фильтр через __init_subclass__ / events

Чтобы не писать where(Post.deleted_at.is_(None)) в каждом запросе, используйте SQLAlchemy event do_orm_execute (SQLAlchemy 1.4+):

from sqlalchemy import event
from sqlalchemy.orm import Session

@event.listens_for(Session, "do_orm_execute")
def filter_soft_deleted(execute_state):
    if (
        not execute_state.is_column_load
        and not execute_state.is_relationship_load
        and not execute_state.execution_options.get("include_deleted", False)
    ):
        execute_state.statement = execute_state.statement.options(
            with_loader_criteria(
                SoftDeleteMixin,
                lambda cls: cls.deleted_at.is_(None),
                include_aliases=True,
            )
        )

Mixin для переиспользования

from sqlalchemy.orm import declared_attr

class SoftDeleteMixin:
    deleted_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True, index=True
    )

    def soft_delete(self) -> None:
        self.deleted_at = datetime.utcnow()

    @classmethod
    def not_deleted(cls):
        return cls.deleted_at.is_(None)


class Post(SoftDeleteMixin, Base):
    __tablename__ = "posts"
    id:    Mapped[int]  = mapped_column(primary_key=True)
    title: Mapped[str]

Запросы с мягким удалением

from sqlalchemy import select

# Только активные записи
stmt = select(Post).where(Post.not_deleted())

# Включить удалённые (с явным флагом)
stmt_all = select(Post).execution_options(include_deleted=True)

# Восстановление
async def restore_post(session: AsyncSession, post_id: int) -> Post:
    post = await session.get(Post, post_id,
                             execution_options={"include_deleted": True})
    post.deleted_at = None
    await session.commit()
    return post

Миграция Alembic

def upgrade():
    op.add_column("posts",
        sa.Column("deleted_at",
                  sa.DateTime(timezone=True),
                  nullable=True))
    op.create_index("ix_posts_deleted_at", "posts", ["deleted_at"])

def downgrade():
    op.drop_index("ix_posts_deleted_at")
    op.drop_column("posts", "deleted_at")

Подводные камни

  • Уникальные индексы по email/slug нужно делать partial: CREATE UNIQUE INDEX ... WHERE deleted_at IS NULL — иначе нельзя создать запись с тем же email после мягкого удаления.
  • Foreign key на мягко удалённую запись остаётся валидным — дочерние записи могут ссылаться на «удалённый» объект, нужны явные проверки.
  • Без глобального фильтра (with_loader_criteria) легко забыть фильтр в одном из запросов и показать удалённые данные пользователю.
  • Backups и аналитика: таблица растёт без VACUUM — старые soft-deleted строки нужно периодически архивировать или физически удалять.
  • ORM-связи (relationship) без primaryjoin фильтра по deleted_at подгружают удалённые дочерние записи.
  • Каскадное мягкое удаление нужно реализовывать вручную — SQLAlchemy cascade не трогает deleted_at.

Common mistakes

  • Описывать soft deletes только как термин и не показывать механизм на минимальном примере.
  • Игнорировать ошибки, пустые данные, конкурентный доступ или границы транзакции.
  • Не связывать поведение с официальным контрактом SQLAlchemy и реальной эксплуатацией.

What the interviewer is testing

  • Объясняет soft deletes через последовательность действий, а не через набор ключевых слов.
  • Приводит короткий кодовый пример или production-сценарий с ожидаемым поведением.
  • Называет хотя бы один риск: производительность, безопасность, транзакции, память или сопровождение.

Sources

Related topics