Обновление и удаление записей

База, в которую можно только добавлять, — это журнал, а не инструмент. Настоящим приложением данные становятся, когда их можно менять: задача выполнена — ставим пометку, опечатка в названии — исправляем, работа закрыта — сносим. Сегодня закрываем весь жизненный цикл записи: SQLAlchemy умеет и менять, и удалять, причём каждым из двух способов — по-разному умным.

Путей два, и выбор между ними — главное решение урока. Путь первый ручной: нашёл объект через select, поменял ему атрибут, вызвал commit — а сессия сама догадалась построить UPDATE. Путь второй массовый: один запрос update() или delete() с условием where проходит по всем подходящим строкам прямо в базе, и Python не видит ни одной из них. Первый путь — про работу с одной записью, второй — про операции над множествами.

Путь 1: нашёл, изменил, commit

Начнём с ручного пути, потому что он объясняет, зачем сессия вообще существует. Когда select возвращает объект, сессия не забывает о нём после выдачи — она продолжает следить за его атрибутами. Поменял поле — сессия это заметила; вызвал commit — она построила UPDATE только из изменившихся колонок. Ни одного слова SQL ты при этом не написал.

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

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

    def __repr__(self):
        return f"Task({self.id}, {self.title!r}, done={self.done})"

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="купить хлеб"), Task(title="полить цветы")])
    s.commit()

    task = s.scalars(select(Task).where(Task.title == "купить хлеб")).one()
    task.done = 1          # просто меняем атрибут - как у обычного объекта
    s.commit()             # сессия сама построит UPDATE

    fresh = s.scalars(select(Task).where(Task.title == "купить хлеб")).one()
    print("стало:", fresh)

стало: Task(1, 'купить хлеб', done=1)

Строка task.done = 1 неотличима от присваивания обычному Python-объекту — и в этом весь смысл ORM. Вся магия происходит на s.commit(): сессия сравнивает текущие значения атрибутов с теми, что были при загрузке, находит изменившиеся и отправляет в базу UPDATE ровно по ним. Такой снимок «как было при загрузке» сессия хранит у каждого отслеживаемого объекта, поэтому изменить можно хоть пять полей сразу — в UPDATE попадут все пять, а нетронутые база не услышит.

Отслеживание работает и в обратную сторону: повторный select того же объекта не читает базу заново и не строит второй объект. Сессия держит карту объектов, которые уже есть в работе, и выдаёт из неё: один объект — один раз на всю сессию, сколько бы запросов его ни достал.

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

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="задача А"), Task(title="задача Б")])
    s.commit()

    a = s.scalars(select(Task).where(Task.title == "задача А")).one()
    again = s.scalars(select(Task).where(Task.title == "задача А")).one()
    print("повторный select вернул тот же объект:", again is a)

повторный select вернул тот же объект: True

Проверка again is a — не синтаксический трюк, а честное сравнение идентичности: оба имени указывают на один и тот же объект в памяти. Это карта идентичности, сердце сессии: пока сессия жива, у каждой строки базы не больше одного Python-представителя. Именно поэтому ручной путь так удобен — достал объект один раз, а дальше работаешь с ним, как с обычным классом.

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add(Task(title="черновик"))
    s.commit()

with Session(engine) as s:
    task = s.get(Task, 1)
    task.title = "не сохранится"     # изменили, но commit забыли

with Session(engine) as s:
    print("в базе:", s.get(Task, 1).title)

в базе: черновик

Обрати внимание на s.get(Task, 1) — быстрый способ достать объект по первичному ключу, короче, чем select с where. Сначала он заглядывает в карту идентичности, и только если объекта там нет — идёт в базу. Для выборки по ключу это самый короткий и идиоматичный путь в SQLAlchemy 2.0.

А если передумал? Пока commit не вызван, всё обратимо: s.rollback() откатывает сессию к последнему закоммиченному состоянию, и изменённый атрибут молча вернётся к прежнему значению. Это ещё одна услуга отслеживания — сессия помнит снимок «как было» и умеет его восстанавливать. В прикладном коде rollback обычно вызывают при ошибке: транзакция не удалась — откатились, база осталась консистентной, можно пробовать снова. Ручной путь, кстати, этим и силён: между изменением и commit можно прогонять любые проверки на Python — посчитать, сравнить, распечатать — и только потом решать, сохранять ли.

Путь 2: массовое обновление update()

Ручной путь хорош для одной записи или горстки. А теперь представь обратную задачу: закрыть тысячу просроченных задач, поднять всем зарплаты, сбросить пароли всем пользователям. Подход «достать всё, поменять в цикле, закоммитить» протащит тысячу объектов через Python, построит тысячу представителей в памяти и тысячу изменений в сессии — а в базе всё это одна команда UPDATE. Для таких объёмов у SQLAlchemy есть массовое обновление: update() с условием и значениями.

from sqlalchemy import create_engine, String, select, update
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="раз"), Task(title="два"), Task(title="три")])
    s.commit()

    r = s.execute(update(Task).where(Task.done == 0).values(done=1))
    s.commit()

    print("изменено строк:", r.rowcount)
    for t in s.scalars(select(Task).order_by(Task.id)):
        print(t.title, "-", t.done)

изменено строк: 3
раз - 1
два - 1
три - 1

Читается запрос почти как английское предложение: обновить Task, где done равно нулю, установив done в единицу. Условие where — то же самое, что в выборках из прошлого урока, значения — словарь-стиль values(колонка=новое_значение). Результат выполнения несёт rowcount — сколько строк база реально изменила; это бесплатный отчёт о проделанной работе, который в ручном пути пришлось бы считать самому.

Массовое обновление через update() идёт одним SQL-запросом прямо в базе: Python не получает ни одной строки и не строит объекты. Поэтому оно и быстрое: база — единственный, кто читает строки, и ей не нужно тащить их по сети в интерпретатор. Для тысяч строк разница с ручным путём — секунды против минут.

from sqlalchemy import create_engine, String, update
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="а"), Task(title="б"), Task(title="в"), Task(title="г")])
    s.commit()

    r = s.execute(update(Task).values(done=1))    # where нет - поменяются все
    s.commit()
    print("where нет - задето строк:", r.rowcount)

where нет - задето строк: 4

from sqlalchemy import create_engine, String, select, update
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as first:
    first.add_all([Task(title="отчёт"), Task(title="аудит")])
    first.commit()
    task = first.scalars(select(Task).where(Task.title == "отчёт")).one()

    with Session(engine) as second:
        second.execute(update(Task).values(done=1))
        second.commit()
        print("вторая сессия:", second.get(Task, 1).done)

    print("первая сессия:", task.done)

вторая сессия: 1
первая сессия: 0

Смотри, что произошло: вторая сессия пометила задачи выполненными и честно сохранила это в базу, а объект в первой сессии так и держит done = 0 — его никто не предупредил. Это не баг, а плата за кэш: сессия верит своим сохранённым объектам больше, чем базе, иначе кэширование не имело бы смысла. В демо-приложениях с одной короткой сессией на запрос проблемы нет; в долгоживущих сервисах про эту асимметрию нужно помнить.

Как выбрать путь в реальном коде? Спроси себя: мне нужен сам объект или только результат в базе? Если перед сохранением надо посчитать, проверить или что-то распечатать — это ручной путь, объект уже в руках. Если же операция формулируется как «всем, у кого условие, поменяй поле» — это массовый путь, и тащить строки в Python ради них не нужно. Ошибка новичка — противоположные крайности: массово менять по одному объекту в цикле из тысячи итераций или писать update() там, где нужна логика приложения. Между этими крайностями и лежит таблица ниже.

Удаление: delete()

Удаление устроено зеркально обновлению — те же два пути. Массовый путь: delete() с условием where удаляет все подходящие строки одним запросом и возвращает rowcount. Ручной путь: session.delete(объект) помечает конкретного представителя к удалению, а commit отправляет DELETE.

from sqlalchemy import create_engine, String, select, delete
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="а", done=1), Task(title="б"), Task(title="в", done=1)])
    s.commit()

    r = s.execute(delete(Task).where(Task.done == 1))
    s.commit()

    print("удалено:", r.rowcount)
    print("осталось:", [t.title for t in s.scalars(select(Task).order_by(Task.id)).all()])

удалено: 2
осталось: ['б']

Условие where здесь работает как в select: только выполненные задачи исчезли, невыполненная осталась. Типовой сценарий — чистка хвостов: удалить все записи старше срока, отвязать помеченные на удаление, вычистить выполненное перед архивацией. Rowcount снова говорит, сколько строк база забрала.

Внутри where работают все операторы из прошлого урока: in_ для списка значений, сравнения, цепочки условий через повторный вызов where. Массовые операции поэтому и называют условными: одно и то же приложение днём удаляет выполненные задачи, а ночью — записи старше месяца, и код отличается только выражением внутри where. Написав условие один раз, его же можно подсунуть и select, и delete: сначала посмотреть список жертв, потом заменить слово select на слово delete.

from sqlalchemy import create_engine, String, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="оставить"), Task(title="выбросить")])
    s.commit()

    victim = s.scalars(select(Task).where(Task.title == "выбросить")).one()
    s.delete(victim)       # пометили к удалению
    s.commit()             # база получила DELETE

    print("осталось:", s.scalar(select(func.count()).select_from(Task)))

осталось: 1

Заодно в этом примере появился func.count() — способ спросить у базы «сколько строк» без их загрузки; подробно его разберём в уроке про агрегации. Пока достаточно заметить: подсчёт тоже можно доверить базе, и это дешевле, чем len() по загруженному списку.

from sqlalchemy import create_engine, String, select, delete, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="одна"), Task(title="две"), Task(title="три")])
    s.commit()

    r = s.execute(delete(Task))     # условия нет - удалятся все строки
    s.commit()
    print("осталось:", s.scalar(select(func.count()).select_from(Task)))

осталось: 0

Что дальше

Теперь каждая запись под полным контролем: изменить по одному — атрибутом и commit, изменить пачкой — update() с where, снести пачкой — delete() с where. Каждая запись живёт своей жизнью, но настоящий менеджер задач — это не свалка строк, а связь: у каждой задачи есть автор, а у автора — его задачи. Как описать такую связь колонкой ForeignKey и научиться ходить от автора к задачам без единого ручного запроса — тема следующего урока.

Ручной путь меняет объект и доверяет сессии построить SQL; массовый путь меняет строки в базе, не поднимая их в Python. Один — для точной работы, другой — для объёмов.