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-сценарий с ожидаемым поведением.
- Называет хотя бы один риск: производительность, безопасность, транзакции, память или сопровождение.