Version columns in SQLAlchemy 2.0: the native implementation

SQLAlchemy 2.0 supports optimistic locking natively. You don't have to reinvent the pattern — add a version column, configure __mapper_args__, and SQLAlchemy takes care of the rest: it increments the counter on every UPDATE, validates that the current version matches at write time, and raises StaleDataError if somebody else updated in the meantime.

In this capsule you're going to see the native implementation, the two variants (an integer counter and a timestamp), and the few details where you have to make decisions (the relationship with migrations, the behavior in bulk updates, what to do with FastAPI). By the end you'll have the canonical pattern you'll reuse in every model where you want optimistic locking.

The basic pattern: an integer counter

# models.py
from sqlalchemy import Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    status: Mapped[str] = mapped_column(String(50), default="pending")
    version: Mapped[int] = mapped_column(default=1)

    __mapper_args__ = {
        "version_id_col": "version",
    }

With this, SQLAlchemy:

1. Increments version automatically on every UPDATE.
2. Includes WHERE version = ? in the generated UPDATE.
3. Raises sqlalchemy.orm.exc.StaleDataError if the UPDATE didn't affect the expected row.

Demonstration

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm.exc import StaleDataError


engine = create_async_engine("postgresql+asyncpg://...")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def demo():
    # Client A reads the task
    async with SessionLocal() as session_a:
        task_a = await session_a.get(Task, 1)
        print(f"Client A read: version={task_a.version}, title={task_a.title}")

    # Client B reads the same task
    async with SessionLocal() as session_b:
        task_b = await session_b.get(Task, 1)
        print(f"Client B read: version={task_b.version}")

        # Client B modifies and saves first
        task_b.title = "Modified by B"
        await session_b.commit()
        print(f"Client B saved: new version={task_b.version}")

    # Client A tries to save — its version is old
    async with SessionLocal() as session_a:
        # Re-attach (in a real case it would be in another request)
        task_a = await session_a.merge(task_a)
        task_a.title = "Modified by A"
        try:
            await session_a.commit()
            print("Client A saved (this shouldn't happen!)")
        except StaleDataError as e:
            print(f"Client A got a StaleDataError: {e}")


asyncio.run(demo())

Output:

Client A read: version=1, title=Original
Client B read: version=1
Client B saved: new version=2
Client A got a StaleDataError: UPDATE statement on table 'tasks'
expected to update 1 row(s); 0 were matched.

The important part: the StaleDataError comes automatically, with no manual version-checking code.

What SQL SQLAlchemy generates

With the configuration above, a typical UPDATE:

task = await session.get(Task, 1)  # version=5
task.title = "New title"
await session.commit()

generates this SQL:

UPDATE tasks
SET title = 'New title', version = 6
WHERE id = 1 AND version = 5;

If that query affects no rows (because another UPDATE changed the version to 6 in the meantime), rowcount = 0 and SQLAlchemy raises StaleDataError.

A migration to add version to an existing table

If you already have the table in production, you need a migration:

# Alembic migration
def upgrade():
    op.add_column(
        'tasks',
        sa.Column('version', sa.Integer(), nullable=False, server_default='1')
    )
    # Remove the default after the backfill (optional)
    op.alter_column('tasks', 'version', server_default=None)


def downgrade():
    op.drop_column('tasks', 'version')

server_default='1' ensures the existing rows have version=1. Afterward you can remove the default if you prefer version to be immutable at the schema level (the app always provides it).

For the migration in production with zero downtime (module 5), adding version with a server_default is expand-compatible: the column exists with a default value, but the old code can insert without mentioning version (PostgreSQL applies the default).

Variant: a timestamp instead of a counter

Sometimes you prefer a timestamp as the version. SQLAlchemy supports this with version_id_generator:

from datetime import datetime, timezone

class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    last_modified: Mapped[datetime] = mapped_column(
        default=lambda: datetime.now(timezone.utc)
    )

    __mapper_args__ = {
        "version_id_col": "last_modified",
        "version_id_generator": lambda v: datetime.now(timezone.utc),
    }

version_id_generator is a function that generates the next value. Here it returns now(UTC) each time.

Counter vs timestamp: the trade-offs

The counter is the default option — simpler, less subtle, with no clock-skew problems. The timestamp only wins if you need to order versions chronologically or use the value for an explicit "last write wins".

Watch out for clock skew

If two servers write to the same table and their clocks differ by seconds (drift is real in distributed systems), the "most recent" one can be wrong. A counter doesn't have that problem — an UPDATE increments monotonically no matter which server it comes from.

If you use a timestamp, configure NTP correctly on all the servers and consider using PostgreSQL's clock_timestamp() (server-side) instead of timestamps from the app.

Behavior in special cases

Bulk updates with update()

# A bulk update: update all the tasks of a project
await session.execute(
    update(Task)
    .where(Task.project_id == 42)
    .values(status='archived')
)
await session.commit()

Careful: version_id_col does NOT apply to bulk updates. SQLAlchemy doesn't increment version automatically in session.execute(update(...)). You have to do it manually:

await session.execute(
    update(Task)
    .where(Task.project_id == 42)
    .values(status='archived', version=Task.version + 1)
)

Or use synchronize_session='fetch' so SQLAlchemy handles it correctly, but the behavior is still manual for the version increment.

This is by design — bulk updates typically don't need individual optimistic locking. If you need it, consider whether a bulk update is the right operation or whether you should iterate individually.

session.merge() with an unknown version

If you receive partial data from a client that doesn't include the version, merge can behave unexpectedly. Better: always include the version in the client→server round-trip.

# An endpoint that receives an update
class TaskUpdateRequest(BaseModel):
    title: Optional[str] = None
    status: Optional[str] = None
    version: int  # Required — the client sends the version it had

@router.put("/tasks/{task_id}")
async def update_task(
    task_id: int,
    update_data: TaskUpdateRequest,
    db: AsyncSession = Depends(get_db),
):
    task = await db.get(Task, task_id)
    if task.version != update_data.version:
        raise HTTPException(409, ...)  # We cover this in capsule 04

    if update_data.title:
        task.title = update_data.title
    if update_data.status:
        task.status = update_data.status

    try:
        await db.commit()
    except StaleDataError:
        # A race condition between the manual check and the commit
        raise HTTPException(409, ...)

    return task

Insert (no conflict, the initial version)

When creating a new record, the version gets set to the default (1 with a counter, now() with a timestamp). No race is possible — an INSERT is atomic.

new_task = Task(title="New task", status="pending")
db.add(new_task)
await db.commit()
# new_task.version is 1 now

Soft delete with a version

If you have soft deletes (module 2), the "delete" is an UPDATE that sets deleted_at. Like any UPDATE, optimistic locking applies:

task.deleted_at = datetime.now(timezone.utc)
await session.commit()  # Generates an UPDATE with a version check

Behavior with schema migrations

If you add a new column to a table with version_id_col, the existing rows have their version equal to its value before the change. The first UPDATE after the migration will increment it to version + 1 normally.

If you rename the version column (rare but possible), you have to update __mapper_args__ simultaneously with the migration. Do an expand-contract (module 5):

1. Add new_version with the value copied from version.
2. Deploy an app that writes to both.
3. After validating, deploy an app that only reads from new_version.
4. Drop version.

Traps and common mistakes

1. Forgetting __mapper_args__.

Adding the version column isn't enough. SQLAlchemy needs the config to apply the pattern:

# ❌ The column alone does nothing
class Task(Base):
    version: Mapped[int] = mapped_column(default=1)

# ✅ With __mapper_args__
class Task(Base):
    version: Mapped[int] = mapped_column(default=1)

    __mapper_args__ = {"version_id_col": "version"}

2. Using version_id_col in bulk updates expecting automatic behavior.

version_id_col only applies to ORM operations (objects modified and committed). session.execute(update(...)) bypasses it. For bulk with optimistic locking, you handle the version manually.

3. Not adding WHERE version = ? in custom endpoints with raw SQL.

If you bypass the ORM with text("UPDATE tasks SET ..."), you also bypass the version check. If you need raw SQL, the check is manual:

UPDATE tasks
SET title = $1, version = version + 1
WHERE id = $2 AND version = $3;
-- Check the rowcount; if it's 0, there's a conflict

4. Catching StaleDataError and silently ignoring it.

Catch + pass is silent data loss. Always translate it to a clear response for the client (we cover this in capsule 04).

5. Thinking version replaces updated_at.

version is for concurrency control. updated_at is for auditing and display ("Last updated 5 minutes ago"). They serve different purposes. Most tables will have both.

6. Mixing a timestamp and a counter in the same table.

A single column as the version_id_col. If you want both (a counter for concurrency + a timestamp for display), have two separate columns: version (counter) and updated_at (timestamp).

7. Defaulting to version=0 instead of version=1.

The convention: start at 1. A client that sees version=0 might think the field is null or uninitialized. A small detail but it helps.

8. Not documenting the pattern in the API.

If your endpoint requires version in the body (and returns a 409 if it doesn't match), document it in OpenAPI/Swagger. With no documentation, the clients don't know the pattern exists.

Exercise: implement and validate a version column

Setup:

# models.py
from sqlalchemy import String, Integer
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    status: Mapped[str] = mapped_column(String(50), default="pending")
    version: Mapped[int] = mapped_column(default=1)

    __mapper_args__ = {
        "version_id_col": "version",
    }

Step 1: create the table and a couple of tasks.

# script.py
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://localhost/test")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def setup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with SessionLocal() as session:
        session.add(Task(title="Task 1", status="pending"))
        session.add(Task(title="Task 2", status="pending"))
        await session.commit()


asyncio.run(setup())

Step 2: simulate concurrency. Two sessions read, both modify, one commits first.

async def simulate_conflict():
    # Session A
    async with SessionLocal() as session_a:
        task_a = await session_a.get(Task, 1)
        print(f"A: version={task_a.version}, title={task_a.title}")

    # Session B (reads the same one)
    async with SessionLocal() as session_b:
        task_b = await session_b.get(Task, 1)
        print(f"B: version={task_b.version}")

        task_b.title = "Modified by B"
        await session_b.commit()
        print(f"B commit: new version={task_b.version}")

    # Session A tries to commit with the old version
    async with SessionLocal() as session_a:
        task_a_reattached = await session_a.merge(task_a)
        task_a_reattached.title = "Modified by A"
        try:
            await session_a.commit()
            print("A: commit succeeded (something is wrong)")
        except StaleDataError as e:
            print(f"A: StaleDataError caught correctly: {e}")

Question: does the simulation reproduce the conflict?

Step 3: verify the version increments on every UPDATE.

async def verify_increment():
    async with SessionLocal() as session:
        task = await session.get(Task, 2)
        original_version = task.version

        for i in range(3):
            task.title = f"Update {i+1}"
            await session.commit()
            print(f"After update {i+1}: version={task.version}")

        # the version should have incremented 3 times
        assert task.version == original_version + 3

Step 4: verify that a bulk update does NOT increment the version automatically.

async def verify_bulk_doesnt_increment():
    from sqlalchemy import update

    async with SessionLocal() as session:
        # A bulk update
        await session.execute(
            update(Task)
            .where(Task.id == 2)
            .values(status="completed")
        )
        await session.commit()

        # Reload
        task = await session.get(Task, 2)
        # the version did NOT increment automatically
        print(f"After the bulk update: version={task.version}")

Step 5: patch the bulk update so it DOES increment.

async def bulk_update_with_version():
    from sqlalchemy import update

    async with SessionLocal() as session:
        await session.execute(
            update(Task)
            .where(Task.id == 2)
            .values(
                status="archived",
                version=Task.version + 1,
            )
        )
        await session.commit()

        task = await session.get(Task, 2)
        print(f"After the bulk with a manual version: version={task.version}")

A: version=1, title=Task 1
B: version=1
B commit: new version=2
A: StaleDataError caught correctly: UPDATE statement on table 'tasks' expected to update 1 row(s); 0 were matched.

After update 1: version=2
After update 2: version=3
After update 3: version=4

After the bulk update: version=2  (it didn't change since the last individual UPDATE)

After the bulk with a manual version: version=3

Summary and next step

What you learned:

• Native optimistic locking in SQLAlchemy 2.0: add a version column + __mapper_args__ = {"version_id_col": "version"}.
• SQLAlchemy automatically: includes WHERE version = ? in UPDATEs, increments the version, raises StaleDataError on a conflict.
• An integer counter is the default. A timestamp is the alternative with version_id_generator — useful for "last write wins", risky with clock skew.
• Bulk updates (session.execute(update(...))) do NOT use version_id_col automatically — you handle the version manually.
• Migration: add the column with server_default='1' for zero downtime; remove the default afterward if you prefer.
• Traps: forgetting __mapper_args__, using bulk without handling the version, raw SQL with no check, silently catching the error.

Before moving on, you should be able to:

• Implement a version column in any SQLAlchemy model.
• Reproduce a StaleDataError with two concurrent sessions.
• Generate the correct migration to add a version to an existing table.
• Handle bulk updates consciously (manually or by avoiding bulk).

In the next capsule we go to the HTTP translation: when we catch StaleDataError, what do we respond to the client? The answer isn't just 409 Conflict — it has to include rich information so the client can resolve the conflict instead of just failing. You're going to see the canonical format of the response body, examples of UX that use that information, and how to distinguish between "fatal" and "resolvable" conflicts where applicable.

Resources

1. SQLAlchemy 2.0 — Versioning Counter — the complete official reference.
2. SQLAlchemy 2.0 — version_id_generator — custom generators.
3. StaleDataError API — the exception's reference.
4. Vlad Mihalcea — JPA Optimistic Locking — the equivalent pattern in another stack, universal principles.
5. Alembic Docs — Adding columns — the reference for the migration.
6. PostgreSQL — Concurrency control — the theoretical base of MVCC that makes all this possible.

Capsule 03 of 08 — Module 6 — SQL Patterns for Production APIs Guide