Module 6: Optimistic Locking + Schema Versioning

Mini-project: TaskFlow with optimistic locking + 3 schema versions

Next →

You close the module by applying everything you learned in a real app. TaskFlow is a task-management API you're going to build with two production-ready features: optimistic locking via the If-Match header + a version column, and schema versioning with three evolving versions — v1 base, v2 adds priority (compatible), v3 gradually deprecates legacy_status (keeping backward-compat).

By the end you have a demo app that demonstrates:

  1. Two concurrent clients editing the same task — the second gets a 412 with rich info to resolve it.
  2. A client with the v1 schema (installed months ago) consuming the app with the v3 schema without breaking — new fields ignored, deprecated fields still present.
  3. Deprecation and Sunset headers correctly set on the v3 endpoints.
  4. Automated tests that verify that clients of each version work.

This is the canonical pattern you're going to reuse every time you build an endpoint with writes and a schema that can evolve. And it's a deliverable that demonstrates mastery in code reviews and interviews.

Project structure

taskflow/
├── README.md
├── docker-compose.yml
├── requirements.txt
├── alembic.ini
├── alembic/
│   └── versions/
│       ├── 001_initial.py
│       ├── 002_add_priority.py
│       └── 003_add_due_at_deprecate_legacy_status.py
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── v1.py
│   │   ├── v2.py
│   │   └── v3.py
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── tasks_v1.py
│   │   ├── tasks_v2.py
│   │   └── tasks_v3.py
│   └── middleware.py
└── tests/
    ├── test_optimistic_locking.py
    ├── test_schema_v1_compat.py
    ├── test_schema_v2_compat.py
    └── test_schema_v3_deprecation.py

The model and the migrations

The final model (after the 3 migrations)

# app/models.py
from datetime import date, datetime, timezone
from typing import Optional

from sqlalchemy import String, Integer, Date, DateTime
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")
    legacy_status: Mapped[Optional[str]] = mapped_column(String(50), nullable=True)  # Deprecated in v3
    priority: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)  # Added in v2
    due_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True)
    due_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)  # Added in v3
    version: Mapped[int] = mapped_column(default=1)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc)
    )

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

Migration 001: the base schema

# alembic/versions/001_initial.py
from alembic import op
import sqlalchemy as sa


def upgrade():
    op.create_table(
        'tasks',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('title', sa.String(200), nullable=False),
        sa.Column('status', sa.String(50), nullable=False, server_default='pending'),
        sa.Column('legacy_status', sa.String(50), nullable=True),
        sa.Column('due_date', sa.Date, nullable=True),
        sa.Column('version', sa.Integer, nullable=False, server_default='1'),
        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
    )


def downgrade():
    op.drop_table('tasks')

Migration 002: add priority (compatible)

# alembic/versions/002_add_priority.py
def upgrade():
    op.add_column(
        'tasks',
        sa.Column('priority', sa.Integer, nullable=True)
    )


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

Migration 003: add due_at, deprecate legacy_status

# alembic/versions/003_add_due_at_deprecate_legacy_status.py
def upgrade():
    op.add_column(
        'tasks',
        sa.Column('due_at', sa.DateTime(timezone=True), nullable=True)
    )
    # legacy_status stays nullable — the deprecation is at the API level, not the DB
    # Backfill: copy legacy_status → status if status is null (defensive)
    op.execute("""
        UPDATE tasks
        SET status = legacy_status
        WHERE status IS NULL AND legacy_status IS NOT NULL
    """)


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

The Pydantic schemas per version

v1 (the original schema)

# app/schemas/v1.py
from datetime import date, datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict


class TaskBase(BaseModel):
    """The v1 base — the original essential fields."""
    model_config = ConfigDict(extra='ignore')  # Forward-compatible

    title: str
    status: str = "pending"
    legacy_status: Optional[str] = None
    due_date: Optional[date] = None


class TaskCreateV1(TaskBase):
    pass


class TaskUpdateV1(BaseModel):
    model_config = ConfigDict(extra='ignore')

    title: Optional[str] = None
    status: Optional[str] = None
    legacy_status: Optional[str] = None
    due_date: Optional[date] = None


class TaskResponseV1(TaskBase):
    id: int
    created_at: datetime

v2 (adds priority — compatible)

# app/schemas/v2.py
from typing import Optional
from app.schemas.v1 import TaskBase, TaskCreateV1, TaskUpdateV1, TaskResponseV1


class TaskCreateV2(TaskCreateV1):
    priority: Optional[int] = None  # New, optional, default None


class TaskUpdateV2(TaskUpdateV1):
    priority: Optional[int] = None


class TaskResponseV2(TaskResponseV1):
    priority: Optional[int] = None  # New in the response

v3 (adds due_at, deprecates legacy_status)

# app/schemas/v3.py
from datetime import datetime
from typing import Optional
from pydantic import ConfigDict, Field
from app.schemas.v2 import TaskCreateV2, TaskUpdateV2, TaskResponseV2


class TaskCreateV3(TaskCreateV2):
    """v3 — adds due_at, keeps legacy_status for backward-compat."""
    due_at: Optional[datetime] = None


class TaskUpdateV3(TaskUpdateV2):
    due_at: Optional[datetime] = None


class TaskResponseV3(TaskResponseV2):
    """v3 — includes due_at, keeps legacy_status (deprecated)."""
    due_at: Optional[datetime] = None
    # legacy_status is still present in the response, inherited from v1
    # but the endpoint marks the field as deprecated in the docs

The v3 endpoint with optimistic locking + deprecation

# app/routers/tasks_v3.py
from datetime import datetime, timezone
from email.utils import formatdate
from typing import Optional

from fastapi import APIRouter, Depends, Header, HTTPException, Response, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.exc import StaleDataError

from app.database import get_db
from app.models import Task
from app.schemas.v3 import TaskCreateV3, TaskUpdateV3, TaskResponseV3


router = APIRouter(prefix="/v3/tasks", tags=["tasks-v3"])


# The sunset date for the legacy_status field
SUNSET_DATE = datetime(2026, 12, 1, 0, 0, 0, tzinfo=timezone.utc)
SUNSET_HEADER = formatdate(SUNSET_DATE.timestamp(), usegmt=True)


def _to_response(task: Task) -> TaskResponseV3:
    return TaskResponseV3(
        id=task.id,
        title=task.title,
        status=task.status,
        legacy_status=task.legacy_status,
        priority=task.priority,
        due_date=task.due_date,
        due_at=task.due_at,
        created_at=task.created_at,
    )


def _add_deprecation_headers_if_needed(task: Task, response: Response):
    """Add Deprecation/Sunset headers if the client is using deprecated fields."""
    if task.legacy_status is not None:
        response.headers["Deprecation"] = "true"
        response.headers["Sunset"] = SUNSET_HEADER
        response.headers["Link"] = (
            '<https://api.example.com/docs/v3-migration>; '
            'rel="deprecation"'
        )


@router.post("", status_code=status.HTTP_201_CREATED)
async def create_task(
    data: TaskCreateV3,
    response: Response,
    db: AsyncSession = Depends(get_db),
) -> TaskResponseV3:
    # Track whether the client is using legacy_status
    if data.legacy_status is not None:
        # In production: a log + a metric
        print(f"[deprecated] legacy_status used in POST /v3/tasks")

    task = Task(
        title=data.title,
        status=data.status,
        legacy_status=data.legacy_status,
        priority=data.priority,
        due_date=data.due_date,
        due_at=data.due_at,
    )
    db.add(task)
    await db.commit()

    response.headers["ETag"] = f'"{task.version}"'
    _add_deprecation_headers_if_needed(task, response)

    return _to_response(task)


@router.get("/{task_id}")
async def get_task(
    task_id: int,
    response: Response,
    db: AsyncSession = Depends(get_db),
    if_none_match: Optional[str] = Header(None, alias="If-None-Match"),
) -> Optional[TaskResponseV3]:
    task = await db.get(Task, task_id)
    if not task:
        raise HTTPException(404, "Task not found")

    current_etag = f'"{task.version}"'

    # If-None-Match for caching
    if if_none_match == current_etag:
        response.status_code = status.HTTP_304_NOT_MODIFIED
        response.headers["ETag"] = current_etag
        return None

    response.headers["ETag"] = current_etag
    _add_deprecation_headers_if_needed(task, response)

    return _to_response(task)


@router.put("/{task_id}")
async def update_task(
    task_id: int,
    data: TaskUpdateV3,
    response: Response,
    db: AsyncSession = Depends(get_db),
    if_match: Optional[str] = Header(None, alias="If-Match"),
) -> TaskResponseV3:
    if not if_match:
        raise HTTPException(
            status.HTTP_428_PRECONDITION_REQUIRED,
            detail="If-Match header required for updates"
        )

    try:
        requested_version = int(if_match.strip('"'))
    except ValueError:
        raise HTTPException(400, "Invalid If-Match header format")

    task = await db.get(Task, task_id)
    if not task:
        raise HTTPException(404, "Task not found")

    # Capture the original state for the diff in case of a conflict
    original_state = {
        "title": task.title,
        "status": task.status,
        "legacy_status": task.legacy_status,
        "priority": task.priority,
        "due_date": task.due_date.isoformat() if task.due_date else None,
        "due_at": task.due_at.isoformat() if task.due_at else None,
    }

    # The pre-check
    if task.version != requested_version:
        raise HTTPException(
            status.HTTP_412_PRECONDITION_FAILED,
            detail={
                "error": "version_mismatch",
                "current_etag": f'"{task.version}"',
                "your_etag": if_match,
                "current_state": _to_response(task).model_dump(mode='json'),
            }
        )

    # Track deprecated field usage
    changes = data.model_dump(exclude_unset=True)
    if "legacy_status" in changes:
        print(f"[deprecated] legacy_status used in PUT /v3/tasks/{task_id}")

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

    try:
        await db.commit()
    except StaleDataError:
        # A race between the check and the commit
        await db.rollback()
        current_task = await db.get(Task, task_id)
        raise HTTPException(
            status.HTTP_412_PRECONDITION_FAILED,
            detail={
                "error": "version_mismatch_race",
                "current_etag": f'"{current_task.version}"',
                "current_state": _to_response(current_task).model_dump(mode='json'),
                "your_changes": changes,
            }
        )

    response.headers["ETag"] = f'"{task.version}"'
    _add_deprecation_headers_if_needed(task, response)

    return _to_response(task)

Optimistic locking tests

# tests/test_optimistic_locking.py
import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_concurrent_updates_second_fails_with_412(client: AsyncClient):
    # Create a task
    create_response = await client.post("/v3/tasks", json={
        "title": "Concurrent test",
        "status": "pending"
    })
    assert create_response.status_code == 201
    task_id = create_response.json()["id"]
    initial_etag = create_response.headers["ETag"]

    # Client A reads the task
    get_a = await client.get(f"/v3/tasks/{task_id}")
    etag_a = get_a.headers["ETag"]

    # Client B also reads it (the same etag)
    get_b = await client.get(f"/v3/tasks/{task_id}")
    etag_b = get_b.headers["ETag"]
    assert etag_a == etag_b

    # Client A updates first
    update_a = await client.put(
        f"/v3/tasks/{task_id}",
        json={"title": "Updated by A"},
        headers={"If-Match": etag_a},
    )
    assert update_a.status_code == 200

    # Client B tries with the old etag
    update_b = await client.put(
        f"/v3/tasks/{task_id}",
        json={"title": "Updated by B"},
        headers={"If-Match": etag_b},
    )
    assert update_b.status_code == 412
    assert update_b.json()["detail"]["error"] == "version_mismatch"
    assert update_b.json()["detail"]["current_etag"] == update_a.headers["ETag"]


@pytest.mark.asyncio
async def test_update_without_if_match_returns_428(client: AsyncClient):
    create = await client.post("/v3/tasks", json={"title": "T"})
    task_id = create.json()["id"]

    # With no If-Match
    response = await client.put(
        f"/v3/tasks/{task_id}",
        json={"title": "Updated"}
    )
    assert response.status_code == 428


@pytest.mark.asyncio
async def test_get_with_if_none_match_returns_304(client: AsyncClient):
    create = await client.post("/v3/tasks", json={"title": "T"})
    task_id = create.json()["id"]
    etag = create.headers["ETag"]

    # A GET with the current etag
    response = await client.get(
        f"/v3/tasks/{task_id}",
        headers={"If-None-Match": etag}
    )
    assert response.status_code == 304

Multi-version compatibility tests

A v1 client consuming a v3 endpoint

# tests/test_schema_v1_compat.py
import pytest
from pydantic import BaseModel, ConfigDict
from httpx import AsyncClient


# A simulation of a v1 client that only knows the v1 fields
class TaskResponseV1Client(BaseModel):
    """The schema a client with the v1 SDK would have."""
    model_config = ConfigDict(extra='ignore')  # Forward-compatible

    id: int
    title: str
    status: str
    legacy_status: str | None = None
    due_date: str | None = None


@pytest.mark.asyncio
async def test_v1_client_consumes_v3_endpoint_successfully(client: AsyncClient):
    # The v3 server creates a task with v3 fields (priority, due_at)
    create = await client.post("/v3/tasks", json={
        "title": "Test",
        "status": "pending",
        "priority": 5,
        "due_at": "2026-12-01T00:00:00Z",
    })
    assert create.status_code == 201
    task_id = create.json()["id"]

    # The v1 client consumes the v3 endpoint
    get_response = await client.get(f"/v3/tasks/{task_id}")
    response_body = get_response.json()

    # The v1 client parses with its schema (which ignores the new fields)
    task_v1 = TaskResponseV1Client(**response_body)

    # It works — the v1 client sees the fields it knows
    assert task_v1.id == task_id
    assert task_v1.title == "Test"
    assert task_v1.status == "pending"
    # The v1 client does NOT crash even though the response has priority and due_at


@pytest.mark.asyncio
async def test_v1_client_ignores_new_v3_fields(client: AsyncClient):
    """Verify that extra='ignore' works correctly."""
    create = await client.post("/v3/tasks", json={
        "title": "T",
        "priority": 3,
        "due_at": "2026-06-01T00:00:00Z",
    })
    response_body = create.json()

    # These fields are in the v3 response
    assert "priority" in response_body
    assert "due_at" in response_body

    # But the v1 client ignores them
    task_v1 = TaskResponseV1Client(**response_body)
    assert not hasattr(task_v1, "priority")
    assert not hasattr(task_v1, "due_at")

The deprecation headers working

# tests/test_schema_v3_deprecation.py
import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_deprecation_headers_set_when_using_legacy_status(client: AsyncClient):
    # Create a task using legacy_status
    create = await client.post("/v3/tasks", json={
        "title": "Test",
        "legacy_status": "open",  # a deprecated field
    })
    assert create.status_code == 201
    assert create.headers.get("Deprecation") == "true"
    assert "Sunset" in create.headers
    assert "deprecation" in create.headers.get("Link", "")


@pytest.mark.asyncio
async def test_no_deprecation_headers_when_not_using_legacy_status(client: AsyncClient):
    # With no legacy_status
    create = await client.post("/v3/tasks", json={
        "title": "Modern task",
        "status": "pending",
    })
    assert create.status_code == 201
    assert "Deprecation" not in create.headers


@pytest.mark.asyncio
async def test_legacy_status_still_works_for_backward_compat(client: AsyncClient):
    # An old client sends legacy_status — it has to work but with the header
    create = await client.post("/v3/tasks", json={
        "title": "Legacy client",
        "legacy_status": "in_progress",
    })
    assert create.status_code == 201
    body = create.json()
    assert body["legacy_status"] == "in_progress"
    assert create.headers.get("Deprecation") == "true"

The BENCHMARKS.md and the deliverables

# TaskFlow — Optimistic Locking + Schema Versioning

A demo API that demonstrates the patterns of:
1. Optimistic locking with `If-Match`/`412`
2. Evolutionary schema versioning (v1 → v2 → v3)
3. Gradual deprecation with `Deprecation`/`Sunset` headers

## Setup

```bash
docker-compose up -d
alembic upgrade head
uvicorn app.main:app --reload

Tests

pytest tests/ -v
# tests/test_optimistic_locking.py: 3 passed
# tests/test_schema_v1_compat.py: 2 passed
# tests/test_schema_v3_deprecation.py: 3 passed

Demo scenarios

Scenario 1: A concurrent edit (optimistic locking)

# Browser A
curl -i http://localhost:8000/v3/tasks/1
# ETag: "5"

# Browser B
curl -i http://localhost:8000/v3/tasks/1
# ETag: "5" (the same)

# A updates first
curl -i -X PUT http://localhost:8000/v3/tasks/1 \
  -H 'If-Match: "5"' \
  -d '{"title": "A"}'
# 200 OK, ETag: "6"

# B tries with the old etag
curl -i -X PUT http://localhost:8000/v3/tasks/1 \
  -H 'If-Match: "5"' \
  -d '{"title": "B"}'
# 412 Precondition Failed
# {"error": "version_mismatch", "current_etag": "\"6\"", ...}

Scenario 2: A v1 client consuming v3

Without installing anything new in the v1 client, it just keeps consuming:

  • It receives extra fields (priority, due_at) that it ignores.
  • It keeps working with the v1 fields (title, status, due_date, legacy_status).

Scenario 3: Gradual deprecation

If the client uses legacy_status:

curl -i http://localhost:8000/v3/tasks/1
# If the task has legacy_status:
# Deprecation: true
# Sunset: Tue, 01 Dec 2026 00:00:00 GMT
# Link: <https://...>; rel="deprecation"

Key takeaways

  1. If-Match + 412 is an HTTP standard, it fits with caches and standard tooling.
  2. extra='ignore' in Pydantic clients enables forward-compatibility automatically.
  3. Gradual deprecation with metrics + headers + communication is the only responsible way to remove features.
  4. A single API version (v3) can serve v1, v2, and v3 clients simultaneously.

---

## Module wrap-up

By completing this project you have:

- A public repo (upload it to GitHub) with optimistic locking + schema versioning code.
- Automated tests that verify concurrency and compatibility.
- A `BENCHMARKS.md` with reproducible demo scenarios.
- A mental pattern to apply in any real API.

What you learned in module 6:

- **Capsule 02:** The optimistic vs pessimistic decision with a matrix.
- **Capsule 03:** The native SQLAlchemy `version_id_col` implementation.
- **Capsule 04:** `StaleDataError` → a `409` response with rich info.
- **Capsule 05:** The `If-Match` header + `412 Precondition Failed`.
- **Capsule 06:** Backward-compatible vs breaking changes.
- **Capsule 07:** The deprecation strategy with `Deprecation`/`Sunset` + metrics.
- **Capsule 08:** The integrative project combining everything.

Before moving on to module 7 you should be able to:

- Implement optimistic locking end-to-end in any mutating endpoint.
- Design an evolution path for an API that's going to live for years.
- Argue against `/v2/` when somebody proposes it.
- Apply gradual deprecation with the right headers and tracking.

---

## We start in the next module

**Module 7** closes the patterns before the complete final project. Topic: **Bulk Operations**. Until now we've talked about "normal" queries — one row at a time, one concurrent UPDATE. But there are cases where you need to insert 100k rows at once (imports, ETL, backfills). You're going to learn PostgreSQL's patterns for doing it at real speed (`COPY`, batch inserts), atomic upserts with `ON CONFLICT`, and when to go bulk vs when to iterate individually.

Before moving on:

- Upload the TaskFlow repo to public GitHub.
- Link it in your portfolio.
- Document the `README.md` well with reproducible instructions.

---

## Resources

1. [SQLAlchemy 2.0 — Versioning](https://docs.sqlalchemy.org/en/20/orm/versioning.html) — the official reference.
2. [RFC 7232 — Conditional Requests](https://datatracker.ietf.org/doc/html/rfc7232) — `If-Match` and ETags.
3. [RFC 8594 — Sunset](https://datatracker.ietf.org/doc/html/rfc8594) — the deprecation header.
4. [Stripe API Versioning](https://stripe.com/blog/api-versioning) — a real case at scale.
5. [Pydantic v2 — `model_config`](https://docs.pydantic.dev/latest/concepts/config/) — `extra='ignore'`.
6. [GitHub Templates — REST API repo](https://github.com/othneildrew/Best-README-Template) — README templates for a portfolio.
7. [HTTPX — async testing](https://www.python-httpx.org/async/) — the library used in the tests.

---

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

*End of module 6. Continue with module 7 for Bulk Operations.*
← PreviousDeprecation strategy: headers, metrics, communicationNext →Module 7: Bulk Operations