Архивные записи вернулись через relationship: with_loader_criteria

Каталог задач фильтрует архивные строки, но экран проекта читает project.tasks и снова показывает архив. Условие Task.archived=False написано в одном endpoint, тогда как другой путь загружает ту же сущность через отношение. При мягком удалении нужно проследить не только основной SELECT, но и способы появления Task в графе объектов.

Условие относится к сущности

with_loader_criteria добавляет критерий выбранному ORM-классу в рамках запроса и поддерживаемых загрузок отношений. Его удобно использовать вместе с selectinload: запрос проектов и последующая загрузка tasks получают согласованное правило активности. include_aliases=True распространяет критерий и на ORM-алиасы соответствующего класса.

Начнём с явной опции конкретного read-запроса. Это делает место применения видимым. Для общего поведения приложения опцию можно добавлять в обработчике ORM-исполнения, но тогда нужно отдельно определить исключение для административного просмотра архива и проверить все пути. Одной настройки выборки недостаточно, чтобы считать произвольные Core-команды и уже загруженные объекты автоматически отфильтрованными.

Сравним полный и активный набор

Нужен SQLAlchemy. В первой Session список содержит обе задачи. Во второй запрос получает только активную. Дополнительно запросим Task напрямую с той же опцией. Так проверка показывает два способа получения сущности, а не просто корректность одного JOIN.

from sqlalchemy import create_engine, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class Project(Base):
    __tablename__ = "projects"
    id: Mapped[int] = mapped_column(primary_key=True)
    tasks: Mapped[list["Task"]] = relationship()

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    project_id: Mapped[int] = mapped_column(ForeignKey("projects.id"))
    archived: Mapped[bool] = mapped_column(default=False)

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
with Session(engine) as db:
    db.add_all([Project(id=1, tasks=[Task(id=10, archived=False),
                                    Task(id=11, archived=True)]), Project(id=2)])
    db.commit()

from sqlalchemy.orm import selectinload, with_loader_criteria

active = with_loader_criteria(Task, Task.archived.is_(False), include_aliases=True)
with Session(engine) as db:
    project = db.scalars(select(Project).where(Project.id == 1)
                         .options(selectinload(Project.tasks))).one()
    assert sorted(t.id for t in project.tasks) == [10, 11]

with Session(engine) as db:
    projects = db.scalars(select(Project).order_by(Project.id)
                          .options(selectinload(Project.tasks), active)).all()
    result = [{"id": p.id, "tasks": [t.id for t in p.tasks]} for p in projects]
    assert result == [{"id": 1, "tasks": [10]}, {"id": 2, "tasks": []}]
    direct = db.scalars(select(Task).options(active)).all()
    assert [t.id for t in direct] == [10]
    print({"projects": result, "direct_tasks": [t.id for t in direct]})
engine.dispose()

Уже прочитанные данные требуют внимания

Новый критерий не вычищает существующую Python-коллекцию сам по себе. Если Session раньше загрузила архивные записи, следующий экран способен получить прежний экземпляр. Для разных режимов просмотра проще использовать разные короткие единицы работы. Если нужно переиспользовать Session, явно обновляйте состояние и проверяйте, что не теряются локальные изменения.

Административный просмотр архива должен включаться отдельным решением, а не случайным пропуском фильтра. Например, функция запроса может принимать разрешённый режим и строить нужные опции. Не позволяйте произвольному query-параметру клиента выключать общий фильтр без проверки полномочий. Иначе аккуратное распространение условия не спасёт от неверного выбора самого режима.

Soft delete влияет на уникальность и связи. Архивная строка остаётся в таблице и может продолжать занимать уникальное имя. Фильтр чтения этого не меняет. Если продукт разрешает повторное использование имени после архивации, правило нужно выразить в схеме базы подходящим ограничением или индексом. Это отдельное решение от скрытия строки в списке.

Для теста возьмите проект только с архивными задачами и убедитесь, что он остаётся с пустой коллекцией, если именно это требуется экрану. Родительская выборка не обязана исчезать вместе с отфильтрованными детьми. Затем проверьте прямое чтение задачи и административный режим. Такой набор обнаружит неполное внедрение правила на разных путях.

Сокращение запросов при загрузке коллекций разобрано в статье об N+1. Здесь цель — одинаковый состав данных, когда одна сущность появляется напрямую и через relationship.

Продолжить на практике

Оформите чтение активных и архивных записей как явные операции репозитория и проверьте их отдельно. В продвинутом Python есть Repository, Protocol и pytest — полезные инструменты для такого разделения поведения. Откройте программу продвинутого Python.