Using ORM-enabled UPDATE and DELETE statements

Execute update(Account).where(...).values(...) or delete(Session).where(...) through session.execute() — one statement instead of loading every object — and let the default synchronize_session="auto" keep any objects the session already holds up to date. This guide belongs to Core vs ORM architecture decisions.

Quick Answer

Loading rows only to change a column and flush them back is the most common source of avoidable queries in ORM code.

Before — load every row, mutate, flush one UPDATE per object:

import datetime as dt

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Account


async def expire_trials(session: AsyncSession, today: dt.date) -> int:
    accounts = (
        await session.scalars(
            select(Account).where(Account.plan == "trial", Account.trial_ends_on < today)
        )
    ).all()
    for account in accounts:
        account.plan = "expired"
    await session.commit()
    return len(accounts)

After — one ORM-enabled UPDATE:

import datetime as dt

from sqlalchemy import update
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Account


async def expire_trials(session: AsyncSession, today: dt.date) -> int:
    result = await session.execute(
        update(Account)
        .where(Account.plan == "trial", Account.trial_ends_on < today)
        .values(plan="expired")
    )
    await session.commit()
    return result.rowcount

The legacy 1.x spelling was session.query(Account).filter(...).update({"plan": "expired"}, synchronize_session=False). The 2.0 form is a plain update() construct, the same one Core uses, executed by the session so it knows to treat the entity as mapped.

Execution Context & Async Workflow Integration

update(Account) names a mapped class rather than a table, and that is what makes it ORM-enabled. When the session executes it, SQLAlchemy resolves the class to its table, translates attribute expressions to columns, applies ORM-level options such as with_loader_criteria, and — the important part — synchronises the identity map afterwards.

The statement runs inside the session's current transaction, on the session's connection, like any other session.execute(). Nothing is committed until the session commits, and a failure rolls back with the rest of the transaction. Under async that is an ordinary await session.execute(...); there is no separate bulk API to learn.

Synchronisation is what synchronize_session controls. Suppose the session already holds Account 42, loaded earlier in the request, and the bulk update changes its plan in the database. Without synchronisation, account.plan in Python still says "trial". The 2.0 default, "auto", first tries "evaluate" — applying the WHERE criteria to in-memory objects in Python — and if the criteria are too complex to evaluate, falls back to "fetch", which on PostgreSQL adds RETURNING accounts.id to learn exactly which rows matched. Matching objects get the new values; nothing else is touched.

RETURNING can also hand back ORM objects directly, which replaces "update, then select what changed":

from sqlalchemy import update

from shop.models import Account

stmt = (
    update(Account)
    .where(Account.plan == "trial", Account.trial_ends_on < today)
    .values(plan="expired")
    .returning(Account)
)
expired = (await session.scalars(stmt)).all()   # Account objects, current values

Deletes work the same way — delete(CartItem).where(CartItem.updated_at < cutoff) — with one difference worth remembering: objects deleted in the database are marked deleted in the session when they are synchronised, so a later access in the same session behaves as it would after session.delete().

These statements sit squarely between the ORM and Core. They use the ORM's names and identity map, and Core's execution model, which is exactly the hybrid the parent guide on Core vs ORM architecture decisions recommends for write-heavy paths.

Resolving Warnings, Errors & Common Mistakes

The skipped-behaviour rows share a cause: a bulk statement never passes the affected rows through the unit of work, so anything that is implemented as part of the unit of work does not happen. That includes @validates methods, attribute events, mapper events, relationship cascades and version counters. Database-side behaviour — triggers, ON DELETE rules, column server_onupdate — still applies, because the database executes the statement. Python-side onupdate column defaults also still apply, because Core renders them into the UPDATE.

Optimistic locking is the case most worth handling explicitly, because skipping it silently loses concurrent edits. Put the version predicate in the statement and treat zero affected rows as a conflict:

from sqlalchemy import update

from shop.errors import StaleEdit
from shop.models import Product


async def rename_product(session, product_id: int, name: str, expected_version: int) -> None:
    result = await session.execute(
        update(Product)
        .where(Product.id == product_id, Product.version == expected_version)
        .values(name=name, version=Product.version + 1)
    )
    if result.rowcount != 1:
        raise StaleEdit(product_id)

The full pattern is in implementing optimistic locking with version counters.

Advanced: Bulk UPDATE by Primary Key and UPDATE FROM

Two variants cover the cases where every matching row should not get the same value.

When each row gets its own values — prices from a supplier feed, positions after a drag-and-drop reorder — pass a list of dictionaries, each containing the primary key, as the second argument to session.execute(). SQLAlchemy groups them and runs an executemany of UPDATE ... WHERE id = :id:

from sqlalchemy import update
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Product


async def apply_price_feed(session: AsyncSession, feed: list[tuple[int, int]]) -> None:
    await session.execute(
        update(Product),
        [{"id": product_id, "price_cents": price} for product_id, price in feed],
    )
    await session.commit()

This form does not use RETURNING and does not support synchronize_session; objects already in the session are not refreshed, so run it in a session that does not hold those products, or expire them afterwards. For very large feeds, the fastest path is often to load the feed into a temporary table and run a single set-based update — the technique in loading rows with Postgres COPY through asyncpg.

When the new value comes from another table, a correlated WHERE across both tables renders as PostgreSQL's UPDATE ... FROM:

from sqlalchemy import update
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Customer, Order


async def denormalise_customer_tier(session: AsyncSession) -> int:
    stmt = (
        update(Order)
        .where(Order.customer_id == Customer.id, Order.status == "open")
        .values(customer_tier=Customer.tier)
        .execution_options(synchronize_session="fetch")
    )
    result = await session.execute(stmt)
    await session.commit()
    return result.rowcount
# UPDATE orders SET customer_tier=customers.tier FROM customers
# WHERE orders.customer_id = customers.id AND orders.status = 'open'

"fetch" is set explicitly here because criteria spanning two tables cannot be evaluated in Python; "auto" would reach the same decision after trying.

All of these forms compose with the session's transaction, so a feed update, an audit insert and a cache-invalidation row can be written atomically in one commit(). That is the practical advantage over dropping to a raw AsyncConnection for bulk work.

Migrating Query.update() and Query.delete() Calls

Legacy code usually has many Query.update() and Query.delete() calls, and they translate mechanically. The details that change are the argument shapes and the synchronisation default.

# 1.x legacy
session.query(Account).filter(Account.plan == "trial").update(
    {Account.plan: "expired"}, synchronize_session=False
)
session.query(CartItem).filter(CartItem.cart_id == cart_id).delete(synchronize_session="fetch")

# 2.0
from sqlalchemy import delete, update

await session.execute(
    update(Account).where(Account.plan == "trial").values(plan="expired")
)
await session.execute(
    delete(CartItem).where(CartItem.cart_id == cart_id)
)

Three things to check on each converted call. First, the old code frequently passed synchronize_session=False because the old default, "evaluate", raised on complex criteria. The new default, "auto", does not raise, so dropping the argument is usually right and makes stale objects less likely. Keep False only where profiling shows the fetch matters and the session provably holds none of the rows.

Second, Query.update() accepted a dictionary keyed by attributes or strings; values() accepts keyword arguments or the same dictionary. Both work — keyword arguments read better and type-check against the model when using the typing guidance from SQLAlchemy 2.0 type annotations.

Third, Query.update() returned the row count directly; session.execute() returns a CursorResult, and the count is result.rowcount. Code that used the return value as a boolean needs the attribute.

With SQLALCHEMY_WARN_20=1 under 1.4, or future=True sessions, each remaining legacy call emits a RemovedIn20Warning naming the method, which turns the migration into a checklist — the process in fixing RemovedIn20Warning deprecation warnings.

Frequently Asked Questions

Related

• Core vs ORM Architecture Decisions — The parent guide: when to use each layer, and how to mix them.
• How to replace Query.filter() with select().where() — The read-side counterpart of this migration.
• Batch inserting millions of rows with Core execute — The insert-side counterpart for bulk writes.
• Implementing optimistic locking with version counters — Keeping version checks when bypassing the unit of work.