`StaleDataError` → HTTP 409 with an informative body

When SQLAlchemy detects a conflict and raises StaleDataError, your app has to translate that into an HTTP response the client understands. The default response {"detail": "Conflict"} with code 409 is technically correct but useless. The client gets the error and doesn't know what to do: what changed? what's the current version? can it retry? did it lose its data?

The difference between an optimistic-locking endpoint that works in production and one that frustrates users lives in the body of the 409 response. A well-designed body gives the client all the information needed to resolve the conflict — most of the time without losing the user's changes.

In this capsule you're going to learn the canonical format of the 409 response body, how to extract the necessary information from SQLAlchemy, examples of UX that use that information for a "merge-friendly" experience, and how to distinguish between "fatal" conflicts (you lost, retry) and "resolvable" ones (review the diff, choose).

Why 409 alone isn't enough

Imagine you're the user. You've been editing a task for 3 minutes. Click "Save". You get:

{
  "detail": "Conflict"
}

What do you do now? Was your work lost? Do you try again? Did you see something wrong? The frustration grows. Most apps in this state simply overwrite — they lose data but at least it "works".

Now imagine the same situation with a rich response:

{
  "error": "stale_version",
  "message": "This task was modified by another user while you were editing.",
  "your_version": 5,
  "current_version": 7,
  "changed_fields": ["status", "assigned_to"],
  "current_state": {
    "id": 123,
    "title": "Fix critical bug",
    "status": "in_review",
    "assigned_to": "alice@example.com",
    "version": 7,
    "last_modified_by": "bob@example.com",
    "last_modified_at": "2026-05-08T14:32:11Z"
  },
  "your_changes": {
    "title": "Fix critical bug (urgent)"
  }
}

Now the client has everything:

• Your version vs the current one.
• Which fields Bob changed (status, assigned_to) vs the ones you modified (title).
• The resource's complete current state, to show the user.
• Your attempted change.

The frontend can show a UI: "Bob assigned this task to Alice and changed the status. Your change (changing the title to 'urgent') doesn't clash with that. Do you want to apply your change on top of the current state?". Click yes → re-submit with version=7. Zero data lost.

The canonical response format

These are the fields I recommend in any optimistic-locking 409 response:

{
  "error": "stale_version",
  "message": "Human-readable explanation",
  "your_version": <int>,
  "current_version": <int>,
  "changed_fields": ["field1", "field2"],
  "current_state": { /* the complete updated resource */ },
  "your_changes": { /* only the fields the client tried to change */ }
}

error: a machine-readable code. "stale_version" is clear. It lets the client switch logic based on this code (vs other kinds of 409 like "duplicate_key").

message: a human explanation, useful if the response gets shown directly to the user.

your_version and current_version: the numbers so the client knows exactly how many revisions it missed.

changed_fields: which fields changed on the server since the version the client had. This requires comparing the current_state against what the client sent as its base — more complex but very valuable for UX.

current_state: the resource's complete current state. The client can show the user what's there now.

your_changes: what the client tried to modify. Useful if the frontend lost track of the user's input.

Some APIs also add:

• merge_strategy: "manual" or "auto_safe" (the server detected there's no overlap and can auto-resolve if the client confirms).
• retry_after: an indication of how long to wait before retrying.
• last_modified_by: the identifier of the actor who modified it (to show "Bob edited this").

Implementation in FastAPI

from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.exc import StaleDataError
from pydantic import BaseModel


router = APIRouter()


class TaskUpdateRequest(BaseModel):
    title: str | None = None
    status: str | None = None
    assigned_to: str | None = None
    version: int  # The client has to send the version it had


class StaleVersionResponse(BaseModel):
    error: str = "stale_version"
    message: str
    your_version: int
    current_version: int
    changed_fields: list[str]
    current_state: dict
    your_changes: dict


@router.put("/tasks/{task_id}")
async def update_task(
    task_id: int,
    update_data: TaskUpdateRequest,
    db: AsyncSession = Depends(get_db),
):
    # Fetch the current state
    task = await db.get(Task, task_id)
    if not task:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Task not found")

    # Capture the original state BEFORE modifying (for the diff)
    original_state = {
        "title": task.title,
        "status": task.status,
        "assigned_to": task.assigned_to,
    }
    original_version = task.version

    # Determine which fields the client wants to change
    changes_requested = update_data.model_dump(
        exclude={"version"},
        exclude_unset=True,  # only the fields the client sent
    )

    # Apply the changes
    for field, value in changes_requested.items():
        setattr(task, field, value)

    try:
        await db.commit()
    except StaleDataError:
        # Reload the current state from the DB (after the rollback)
        await db.rollback()
        current_task = await db.get(Task, task_id)

        # Compute which fields changed on the server vs the client's base
        changed_fields = []
        for field in original_state:
            if getattr(current_task, field) != original_state[field]:
                changed_fields.append(field)

        response = StaleVersionResponse(
            message="This task was modified by another user. Please review changes.",
            your_version=update_data.version,
            current_version=current_task.version,
            changed_fields=changed_fields,
            current_state={
                "id": current_task.id,
                "title": current_task.title,
                "status": current_task.status,
                "assigned_to": current_task.assigned_to,
                "version": current_task.version,
            },
            your_changes=changes_requested,
        )

        raise HTTPException(
            status.HTTP_409_CONFLICT,
            detail=response.model_dump(),
        )

    return task

Three important details:

1. We capture original_state before modifying. Without it we can't compute changed_fields.
2. exclude_unset=True in model_dump(): the client can send only the fields it wants to change, not all of them. Pydantic distinguishes "not sent" from "sent as null".
3. Reload after the rollback: after a StaleDataError, the session is in a rollback state. You have to re-fetch the current state to include it in the response.

Distinguishing resolvable conflicts from fatal ones

Some conflicts are resolvable: your change doesn't clash with the other user's change. E.g. you changed title, Bob changed assigned_to. Those changes can be combined.

Others are fatal: both modified the same field to different values. There's no automatic way to combine them — somebody has to lose.

Detecting them requires comparing your_changes with changed_fields:

# changed_fields: what changed on the server
# your_changes: what the client tried to change
overlap = set(changed_fields) & set(changes_requested.keys())

if not overlap:
    # No overlap — resolvable
    response.merge_strategy = "auto_safe"
    response.message = (
        "Your changes don't conflict with recent updates. "
        "You can safely apply them on top of the current version."
    )
else:
    # Overlap — fatal or requires review
    response.merge_strategy = "manual"
    response.conflicting_fields = list(overlap)
    response.message = (
        f"Your changes to {overlap} conflict with recent updates. "
        "Please review and decide how to merge."
    )

The frontend can use merge_strategy to offer a different UI:

• auto_safe: an "Apply on top" button. The client re-submits with version=current_version and the same changes.
• manual: a side-by-side comparison UI, the user chooses.

UX patterns for the client

Pattern 1: a simple retry with the new version

For trivial conflicts (rare, the same user in two tabs, no overlap):

async function saveTask(task, retries = 1) {
  try {
    return await api.put(`/tasks/${task.id}`, task);
  } catch (e) {
    if (e.status === 409 && retries > 0) {
      // Re-fetch
      const current = await api.get(`/tasks/${task.id}`);
      // Apply our changes on top of the new version
      const updated = { ...task, version: current.version };
      // Retry
      return await saveTask(updated, retries - 1);
    }
    throw e;
  }
}

It works if conflicts are rare and the user's changes aren't sensitive to external state. Risky with complex changes because it overwrites the other user's work.

Pattern 2: reviewing changes with a dedicated UI

For conflicts where the context matters (collaborative editing):

catch (e) {
  if (e.status === 409) {
    showConflictDialog({
      yourVersion: e.your_version,
      currentVersion: e.current_version,
      changedFields: e.changed_fields,
      currentState: e.current_state,
      yourChanges: e.your_changes,
      onResolve: (resolution) => {
        if (resolution === 'overwrite') {
          // Submit with version=current
          api.put(...);
        } else if (resolution === 'discard') {
          // Load current_state, discard the changes
          loadState(e.current_state);
        } else if (resolution === 'merge') {
          // A field-by-field merge UI
          showMergeUI(...);
        }
      }
    });
  }
}

Pattern 3: optimistic with a local backup

A mobile app that persists changes locally before sending:

saveLocallyAndSync(task) {
  // Save locally with a timestamp
  localStorage.setItem(`task_${task.id}_pending`, JSON.stringify({
    task,
    timestamp: Date.now()
  }));

  // Try to sync
  api.put(...).then(() => {
    localStorage.removeItem(`task_${task.id}_pending`);
  }).catch(e => {
    if (e.status === 409) {
      // Show a "There's a conflict — review?" dialog
      // The change stays in localStorage until the user decides
    }
  });
}

The race condition between the If-Match check and the commit

If you validate the version manually before the commit, there's a race:

# A race condition
task = await db.get(Task, task_id)
if task.version != update_data.version:  # T0: the check
    raise HTTPException(409)
# T1: another UPDATE lands here
task.title = update_data.title
await db.commit()  # T2: SQLAlchemy does its check (we already saw it exists)

Between T0 and T2, another UPDATE can land. Your manual check passes, but SQLAlchemy's check (at commit) fails with StaleDataError.

The solution: trust SQLAlchemy. Don't do a manual check. Catch StaleDataError and return a 409 — that's exactly what it's for.

# ✅ Correct: let SQLAlchemy do the atomic check
task = await db.get(Task, task_id)
# (assign the client's version so SQLAlchemy uses that value in the WHERE)
task.version = update_data.version  # A hack: force the version for the check

# Modify
task.title = update_data.title

try:
    await db.commit()
except StaleDataError:
    # A real conflict, with the atomic check already done by SQLAlchemy
    raise HTTPException(409, ...)

But there's a subtlety: SQLAlchemy compares the in-memory version_id with the one that goes into the WHERE. If you've reassigned it, what matters is what got loaded into memory when they did the get().

A cleaner pattern: use the If-Match header (capsule 05) so the version comes in the headers, not the body, and you don't manipulate SQLAlchemy's internal version.

Traps and common mistakes

1. A 409 with {"detail": "Conflict"} and nothing else.

Useless to the client. Always include rich info.

2. Not doing a rollback before the re-fetch.

After a StaleDataError, the session is in an error state. You have to await db.rollback() before any new query.

3. Returning the DB's current_state with sensitive fields.

If you have private fields (passwords, internal notes, billing), don't expose them in the current_state of the 409 response. Use an explicit Pydantic schema that filters them.

4. Using 200 OK instead of 409.

Some backends "absorb" the conflict and return a 200 with a warning body. This breaks the HTTP contract — clients should be able to trust that 200 means success.

5. Using 400 Bad Request or 500 Internal Server Error.

400 implies the client sent something wrong — false, it sent something valid but outdated. 500 implies a server error — false, it's working as designed. 409 Conflict is the correct code.

6. Computing changed_fields wrong from an incomplete reload.

If you reload with db.get(Task, task_id) after the rollback, the session may return the cached object. Use db.refresh(task) or db.expire(task) first to force a reload.

7. Not considering concurrent reads in the reload.

Between your check and your reload, another UPDATE may have landed. The response's current_state may already be out of date by the time it reaches the client. That's OK — the client knows the state may have changed again, and the flow repeats.

8. Excessive logging of 409s.

StaleDataError is expected behavior, not an error. Logging it as ERROR/WARNING fills the logs with false positives. Log it as INFO or as a separate metric (stale_writes_total).

Exercise: implement a rich 409 response

Setup: use the Task model from the previous capsule with version_id_col.

Step 1: implement the endpoint with a simple 409 response:

@router.put("/tasks/{task_id}")
async def update_task_basic(
    task_id: int,
    update_data: TaskUpdateRequest,
    db: AsyncSession = Depends(get_db),
):
    task = await db.get(Task, task_id)
    # ... apply the changes ...
    try:
        await db.commit()
    except StaleDataError:
        raise HTTPException(409, detail="Conflict")
    return task

Test it with curl:

curl -X PUT http://localhost:8000/tasks/1 \
  -H "Content-Type: application/json" \
  -d '{"title": "New", "version": 1}'
# If it's already at version 2: { "detail": "Conflict" }

Step 2: improve the response with rich info as in the capsule.

# ... the complete code from the "Implementation in FastAPI" section ...

Test again:

curl -X PUT http://localhost:8000/tasks/1 \
  -H "Content-Type: application/json" \
  -d '{"title": "New title", "version": 1}'

# Response:
# {
#   "detail": {
#     "error": "stale_version",
#     "message": "...",
#     "your_version": 1,
#     "current_version": 3,
#     "changed_fields": ["status", "assigned_to"],
#     "current_state": {...},
#     "your_changes": {"title": "New title"}
#   }
# }

Step 3: add overlap detection logic (resolvable vs fatal).

overlap = set(changed_fields) & set(changes_requested.keys())
merge_strategy = "auto_safe" if not overlap else "manual"
# Include it in the response

Step 4: simulate a client that gets a 409 and resolves the conflict.

Implement (in pseudo-code):

async def smart_save(task_id, changes, original_version):
    response = await api.put(f"/tasks/{task_id}", {**changes, "version": original_version})

    if response.status == 409:
        body = response.json()["detail"]

        if body["merge_strategy"] == "auto_safe":
            # Re-submit with the new version
            return await api.put(f"/tasks/{task_id}", {
                **changes,
                "version": body["current_version"]
            })
        else:
            # Show a UI to the user
            return await prompt_user_to_resolve(body)

    return response.json()

Step 5: check the logs. Verify that conflicts aren't logged as ERROR.

async def smart_save(task_id, changes, original_version):
    response = await api.put(...)

    if response.status == 409:
        body = response.json()["detail"]

        if body["merge_strategy"] == "auto_safe":
            # Without asking the user, re-apply
            return await api.put(f"/tasks/{task_id}", {
                **changes,
                "version": body["current_version"]
            })
        # ... else: a manual UI

import logging
logger = logging.getLogger(__name__)

# In the except StaleDataError:
logger.info(
    "Optimistic lock conflict on Task %d (your=%d, current=%d)",
    task_id, update_data.version, current_task.version
)

# And a metric
metrics.increment("api.optimistic_conflicts", tags={"resource": "task"})

Summary and next step

What you learned:

• A 409 Conflict alone isn't enough — the body has to be rich in information.
• The canonical format: error, message, your_version, current_version, changed_fields, current_state, your_changes.
• Distinguishing resolvable from fatal: if the fields changed on the server don't overlap with the client's, it's resolvable.
• Implementation in FastAPI: capture original_state before the UPDATE, catch StaleDataError, do a rollback, reload, compare the fields, return a rich response.
• UX patterns: a simple retry, a review dialog, a local backup. Each has its case.
• Traps: an empty 409, not doing a rollback, exposing sensitive fields, using 400/500/200, log spam.

Before moving on, you should be able to:

• Implement an endpoint with a 409 response with rich information.
• Distinguish a resolvable merge from a manual one based on overlap.
• Catch StaleDataError correctly with a rollback before the reload.
• Design the client's UX for conflicts (auto vs manual).

In the next capsule we take the pattern to the HTTP level: using the If-Match header (RFC 7232) instead of passing version in the body. This fits REST semantics better, allows ETags as the version, and makes your API consumable by standard tools (curl, browsers, caching proxies). We also cover 412 Precondition Failed vs 409 Conflict and when to use each.

Resources

1. MDN — HTTP 409 Conflict — the standard reference.
2. Stripe API — Idempotency and Errors — a real example of rich responses.
3. GitHub API — Conflict handling — an example in optimistic concurrency.
4. Vlad Mihalcea — Handling optimistic locking exceptions — handling patterns.
5. Pydantic — model_dump(exclude_unset=True) — the reference.
6. SQLAlchemy 2.0 — session.refresh — for the reload after a rollback.
7. Martin Fowler — Patterns for Error Handling — general patterns for error responses.

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