Using CTEs with INSERT, UPDATE and DELETE ... RETURNING

Turn a DML statement into a CTE with delete(Order).where(...).returning(*Order.__table__.c).cte("moved"), then read its returned rows in the main statement — for example insert(OrderArchive).from_select(columns, select(moved)) — so a multi-step write runs as one atomic statement. This guide belongs to common table expressions, CTEs and recursive queries.

Quick Answer

Moving rows between tables is the classic case: the usual three-step version copies every row through Python and back.

Three round trips, or one statement Left: select the old orders into Python, insert them into the archive table, then delete them from orders; three statements, rows travel to the application and back, and a crash between steps needs the transaction to clean up. Right: WITH moved AS DELETE FROM orders WHERE placed before cutoff RETURNING the columns, INSERT INTO order_archive SELECT FROM moved; one statement, no rows in Python, atomic by construction. select, insert, delete SELECT old orders → Python INSERT INTO order_archive ... DELETE FROM orders WHERE id IN (...) three round trips, rows copied twice one data-modifying CTE WITH moved AS (DELETE ... RETURNING *) INSERT INTO order_archive SELECT ... FROM moved one round trip, no rows in Python The rows the DELETE returns are exactly the rows it deleted, so nothing can be archived twice or missed.

Before — select, insert, delete:

import datetime as dt

from sqlalchemy import delete, insert, select
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Order, OrderArchive


async def archive_orders(session: AsyncSession, cutoff: dt.date) -> int:
    rows = (await session.execute(
        select(Order.__table__).where(Order.placed_on < cutoff)
    )).mappings().all()
    if rows:
        await session.execute(insert(OrderArchive), [dict(r) for r in rows])
        await session.execute(delete(Order).where(Order.id.in_([r["id"] for r in rows])))
    await session.commit()
    return len(rows)

After — one statement with a data-modifying CTE:

import datetime as dt

from sqlalchemy import delete, insert, select
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Order, OrderArchive

ARCHIVED_COLUMNS = ["id", "customer_id", "status", "total_cents", "placed_on"]


async def archive_orders(session: AsyncSession, cutoff: dt.date) -> int:
    moved = (
        delete(Order)
        .where(Order.placed_on < cutoff)
        .returning(*(Order.__table__.c[name] for name in ARCHIVED_COLUMNS))
        .cte("moved")
    )
    stmt = (
        insert(OrderArchive)
        .from_select(ARCHIVED_COLUMNS, select(*(moved.c[name] for name in ARCHIVED_COLUMNS)))
        .returning(OrderArchive.id)
    )
    archived = (await session.scalars(stmt)).all()
    await session.commit()
    return len(archived)
# WITH moved AS (DELETE FROM orders WHERE orders.placed_on < $1
#                RETURNING orders.id, orders.customer_id, ...)
# INSERT INTO order_archive (id, customer_id, ...) SELECT moved.id, ... FROM moved
# RETURNING order_archive.id

No row passes through Python, the operation cannot archive a row without deleting it or the reverse, and the whole thing is one round trip.

Execution Context & Async Workflow Integration

PostgreSQL allows INSERT, UPDATE and DELETE inside a WITH clause, and a RETURNING list turns their affected rows into a relation the rest of the statement can read. SQLAlchemy exposes this by letting any DML construct with .returning() become a CTE through .cte(name). The resulting object has a .c collection of the returned columns and can be used in select(), joins and from_select() like any other CTE.

One snapshot, RETURNING as the only channel Four steps. The statement takes one snapshot. The DELETE in the CTE removes old orders and emits their rows through RETURNING. The main INSERT reads those returned rows from the CTE, not from the orders table, which in this snapshot still shows the deleted rows. All modifications become visible together when the statement completes. statement starts one snapshot for every part CTEs and main statement alike CTE: DELETE ... RETURNING rows removed, rows emitted the only way to see its effect main: INSERT ... SELECT FROM moved reads the returned rows orders still looks unchanged here statement ends all changes visible together Two parts that modify the same row give unpredictable results. Give each row one writer per statement.

Three rules from PostgreSQL shape how these statements behave.

Every part runs against one snapshot. The CTE and the main statement see the database as it was when the statement started. In the archive example, the main INSERT cannot see that orders rows were deleted — if it selected from orders it would still find them — which is why it reads from moved, the CTE's RETURNING output. RETURNING is the only channel through which one part learns what another did.

All parts execute, exactly once, whether or not anything reads them. A data-modifying CTE runs to completion even if the main statement never references it. That is useful for side effects, and it differs from ordinary SELECT CTEs, which PostgreSQL may skip or inline.

Two parts must not modify the same row. Which modification wins is not defined. Design statements so each row has one writer.

The data-modifying WITH must be at the top level of the statement, and SQLAlchemy's PostgreSQL dialect renders it there — above the INSERT ... SELECT, not inside the SELECT. Under asyncpg the statement is sent as a single prepared statement, and await session.execute() or session.scalars() returns whatever the main statement's RETURNING produces.

Because the statement runs through the session, it shares the session's transaction: the archive, an audit insert and an ORM flush can commit together. What it does not do is update the identity map. Order objects already loaded in this session for archived rows are not marked deleted, so either run archival in a session that has not loaded them, or session.expire_all() afterwards. The same trade-off applies to all set-based writes and is discussed in using ORM-enabled UPDATE and DELETE statements.

Resolving Warnings, Errors & Common Mistakes

Exact error or symptomRoot CauseProduction Fix
WITH clause containing a data-modifying statement must be at the top levelA DML CTE nested inside a subquery rather than attached to the top-level statement.Reference the CTE from the top-level statement, or attach it with add_cte().
A side-effect CTE never runsThe main statement does not reference it, so SQLAlchemy never rendered it.stmt.add_cte(audit_cte).
Main statement does not see rows the CTE insertedEvery part uses one snapshot.Read the CTE's RETURNING output instead of the table.
AttributeError or NoSuchColumnError on cte.c.<name>The column was not in .returning(...).Return every column the main statement needs.
Archived Order objects still appear loaded in the sessionSet-based statements do not touch the identity map.Use a fresh session, or session.expire_all().
ForeignKeyViolationError on deleteChild rows reference the orders being archived.Archive children first in an earlier CTE, or use ON DELETE CASCADE.
Three constructs to know Three tiles. insert, update or delete with returning and then cte turns a DML statement into a named CTE that other parts can select from. from_select on an insert builds INSERT INTO table SELECT from the CTE. add_cte attaches a CTE the main statement never references, so it still executes, which is needed for side-effect-only CTEs such as writing an audit row. .returning(...).cte("moved") DML as a named CTE selectable like a table insert(T).from_select(cols, sel) INSERT ... SELECT reads from the CTE stmt.add_cte(audit) unreferenced CTE still executes PostgreSQL executes a data-modifying CTE even if nothing reads it; SQLAlchemy only renders it if attached.

The side-effect case is the one that surprises people coming from PostgreSQL itself, where an unreferenced data-modifying CTE still executes. SQLAlchemy only renders CTEs it finds while compiling the statement, so a CTE nothing refers to is silently omitted. add_cte() attaches it explicitly:

from sqlalchemy import func, insert, literal, select, update

from shop.models import AuditEvent, Product

repriced = (
    update(Product)
    .where(Product.category_id == 7)
    .values(price_cents=Product.price_cents * 110 / 100)
    .returning(Product.id, Product.price_cents)
    .cte("repriced")
)
audit = (
    insert(AuditEvent)
    .from_select(
        ["entity", "entity_id", "action"],
        select(literal("product"), repriced.c.id, literal("reprice_10pct")),
    )
    .cte("audit")
)
stmt = select(func.count()).select_from(repriced).add_cte(audit)
changed = await session.scalar(stmt)

Here the main statement counts repriced products, the audit CTE writes one audit row per product from the same RETURNING output, and add_cte() makes sure the audit CTE is part of the SQL even though the count never reads it.

Advanced: Chaining Inserts Through Generated Keys

The most useful data-modifying CTEs pass a generated primary key from one insert to the next, which is otherwise a round trip per parent. Converting a cart into an order is a good example: create the order, copy the cart's items into order lines pointing at the new order id, and empty the cart — atomically, in one statement.

Cart to order in one statement Three bands. The first CTE inserts the order row and returns its new id and the cart id. The second CTE inserts order lines by selecting cart items joined to the new order CTE, so each line gets the new order id. The main statement deletes the cart items and returns the new order id to the application. new_order: INSERT INTO orders (...) RETURNING id the generated primary key is available to the rest of the statement lines: INSERT INTO order_lines SELECT new_order.id, item.product_id, ... every cart item becomes a line pointing at the new order main: DELETE FROM cart_items WHERE cart_id = :cart RETURNING ... the cart is emptied in the same atomic statement
from sqlalchemy import delete, func, insert, literal, select
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import CartItem, Order, OrderLine, Product


async def checkout(session: AsyncSession, cart_id: int, customer_id: int) -> int:
    total = (
        select(func.coalesce(func.sum(CartItem.quantity * Product.price_cents), 0))
        .join(Product, Product.id == CartItem.product_id)
        .where(CartItem.cart_id == cart_id)
        .scalar_subquery()
    )
    new_order = (
        insert(Order)
        .values(customer_id=customer_id, status="pending", total_cents=total)
        .returning(Order.id)
        .cte("new_order")
    )
    lines = (
        insert(OrderLine)
        .from_select(
            ["order_id", "product_id", "quantity", "unit_price_cents"],
            select(new_order.c.id, CartItem.product_id, CartItem.quantity, Product.price_cents)
            .join(Product, Product.id == CartItem.product_id)
            .where(CartItem.cart_id == cart_id),
        )
        .cte("lines")
    )
    emptied = (
        delete(CartItem)
        .where(CartItem.cart_id == cart_id)
        .returning(literal(1))
        .cte("emptied")
    )
    stmt = select(new_order.c.id).add_cte(lines).add_cte(emptied)
    order_id = await session.scalar(stmt)
    await session.commit()
    return order_id

The lines insert selects CartItem rows from the snapshot, so it sees the cart exactly as it was when the statement began — before emptied deleted those rows. That is the snapshot rule working in your favour: the copy and the delete cannot disagree about which items were in the cart.

The same technique inserts a parent and many children when the children come from Python rather than a table. Pass the child rows through a values() construct used as a selectable, and join it to the parent CTE. For very large child sets, a staging table loaded with COPY is faster, as in loading rows with Postgres COPY through asyncpg.

Keep an eye on readability. A statement with four chained CTEs is atomic and fast, and it is also the hardest thing in the codebase to debug. Name every CTE for what it holds, keep each one to a single responsibility, and prefer a transaction with two or three ordinary statements when the round trips do not matter.

Batching Large Archive Runs

A single WITH moved AS (DELETE ...) over ten million rows is one transaction that deletes ten million rows, writes ten million archive rows, and generates WAL for both before committing. It holds row locks throughout, bloats both tables until vacuum catches up, and if it fails at the end, all of it rolls back. Large archives run in batches.

One huge statement vs batches Bar chart, illustrative. Longest lock hold: the single statement holds row locks for the whole run, batches for a fraction of a second each. Work lost on failure: the single statement loses everything, a batch loses at most one batch. WAL written before any commit: all of it for the single statement, one batch worth for batches. single statement: longest lock hold the entire run batches of 5,000: longest lock hold one batch single statement: work lost on failure all 10 million rows batches: work lost on failure at most 5,000 rows Illustrative proportions. Batches also let vacuum reclaim space while the run continues.

A batch needs a bounded DELETE, and PostgreSQL's DELETE has no LIMIT. The standard workaround selects a batch of keys with FOR UPDATE SKIP LOCKED in an inner CTE and deletes by those keys:

import datetime as dt

from sqlalchemy import delete, insert, select
from sqlalchemy.ext.asyncio import async_sessionmaker

from shop.models import Order, OrderArchive

COLUMNS = ["id", "customer_id", "status", "total_cents", "placed_on"]


async def archive_in_batches(
    Session: async_sessionmaker, cutoff: dt.date, batch_size: int = 5_000
) -> int:
    total = 0
    while True:
        async with Session() as session:
            batch = (
                select(Order.id)
                .where(Order.placed_on < cutoff)
                .order_by(Order.id)
                .limit(batch_size)
                .with_for_update(skip_locked=True)
                .cte("batch")
            )
            moved = (
                delete(Order)
                .where(Order.id.in_(select(batch.c.id)))
                .returning(*(Order.__table__.c[name] for name in COLUMNS))
                .cte("moved")
            )
            stmt = insert(OrderArchive).from_select(
                COLUMNS, select(*(moved.c[name] for name in COLUMNS))
            ).returning(OrderArchive.id)
            archived = len((await session.scalars(stmt)).all())
            await session.commit()
        total += archived
        if archived < batch_size:
            return total

Each loop iteration is its own session and transaction, so locks are short, WAL is written in manageable pieces, and a failure loses one batch rather than the whole run. SKIP LOCKED lets two archive workers run in parallel without blocking each other, and it also means a row an application transaction is currently updating is simply left for a later batch. Ordering by id keeps batches walking the primary key index, which stays fast as the table shrinks.

Tune the batch size by watching lock wait times and replication lag rather than raw throughput: a batch that is fast on the primary but produces WAL faster than replicas apply it is too large. For the pagination side of this — walking a big table predictably by key — see keyset pagination.

Frequently Asked Questions

Does SQLAlchemy support data-modifying CTEs?

Yes. Any insert(), update() or delete() with .returning() can become a CTE with .cte(name), and the PostgreSQL dialect renders it at the top level of the statement.

Why does my audit CTE not run?

Because nothing in the main statement references it, so SQLAlchemy did not render it. Attach it with stmt.add_cte(audit).

Can the main statement see rows inserted by a CTE?

Not by querying the table: every part of the statement shares one snapshot. It can read them through the CTE, which exposes the RETURNING columns.

Do data-modifying CTEs work on MySQL or SQLite?

No. They are a PostgreSQL feature. On other databases, run the steps as separate statements inside one transaction.