Bulk endpoint + live zero-downtime migration

This capsule is the point of the module. You'll implement the POST /tasks/bulk endpoint with COPY + ON CONFLICT (the last two patterns), and then execute the zero-downtime migration live with wrk running in the background. The binary criterion: 0 errors in 600 seconds of real traffic.

If it passes, your project proves you understand the patterns well enough to apply them integrated without breaking production. It's the criterion that separates "I read about patterns" from "I can apply them".

---

POST /tasks/bulk endpoint

# app/routers/tasks_bulk.py
import time
from datetime import datetime, timezone
import uuid

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession

from app.deps import get_db_with_tenant, get_current_tenant_id


router = APIRouter()


class TaskBulkItem(BaseModel):
    project_id: uuid.UUID
    external_id: str = Field(min_length=1, max_length=100)
    title: str = Field(min_length=1, max_length=200)
    status: str
    priority: int = Field(ge=1, le=5)


class BulkUpsertResponse(BaseModel):
    inserted: int
    updated: int
    skipped: int
    errors: list[dict]
    elapsed_ms: int


VALID_STATUSES = ['pending', 'in_progress', 'completed', 'archived']
MAX_TASKS = 50_000


@router.post("/tasks/bulk", response_model=BulkUpsertResponse)
async def bulk_upsert_tasks(
    tasks: list[TaskBulkItem],
    tenant_id: str = Depends(get_current_tenant_id),
    db: AsyncSession = Depends(get_db_with_tenant),
):
    if len(tasks) == 0:
        raise HTTPException(400, "Empty payload")
    if len(tasks) > MAX_TASKS:
        raise HTTPException(413, f"Max {MAX_TASKS} tasks per request")

    start = time.perf_counter()

    now = datetime.now(timezone.utc)
    tenant_uuid = uuid.UUID(tenant_id)

    records = [
        (
            uuid.uuid4(),  # id
            tenant_uuid,
            t.project_id,
            t.external_id,
            t.title,
            t.status,
            t.priority,
            now,
            now,
            1,  # version
        )
        for t in tasks
    ]

    # Access the raw asyncpg connection
    raw_conn = await db.connection()
    asyncpg_conn = await raw_conn.get_raw_connection()
    pg_conn = asyncpg_conn.driver_connection

    async with pg_conn.transaction():
        # Set context for the triggers
        await pg_conn.execute(f"SET LOCAL app.tenant_id = '{tenant_id}'")

        # Create temp table
        await pg_conn.execute("""
            CREATE TEMP TABLE tmp_tasks_import (
                id UUID,
                tenant_id UUID,
                project_id UUID,
                external_id TEXT,
                title TEXT,
                status TEXT,
                priority INT,
                created_at TIMESTAMPTZ,
                updated_at TIMESTAMPTZ,
                version INT,
                valid BOOL DEFAULT TRUE,
                error_msg TEXT
            ) ON COMMIT DROP
        """)

        # COPY records
        await pg_conn.copy_records_to_table(
            "tmp_tasks_import",
            records=records,
            columns=[
                "id", "tenant_id", "project_id", "external_id",
                "title", "status", "priority",
                "created_at", "updated_at", "version",
            ],
        )

        # Validate in SQL
        await pg_conn.execute(f"""
            UPDATE tmp_tasks_import
            SET valid = FALSE, error_msg = 'invalid status'
            WHERE status NOT IN ({','.join(f"'{s}'" for s in VALID_STATUSES)})
        """)

        # INSERT...SELECT...ON CONFLICT
        upsert_result = await pg_conn.fetch("""
            INSERT INTO tasks (
                id, tenant_id, project_id, external_id,
                title, status, priority, created_at, updated_at, version
            )
            SELECT id, tenant_id, project_id, external_id,
                   title, status, priority, created_at, updated_at, version
            FROM tmp_tasks_import WHERE valid = TRUE
            ON CONFLICT (tenant_id, external_id) DO UPDATE SET
                title = EXCLUDED.title,
                status = EXCLUDED.status,
                priority = EXCLUDED.priority,
                updated_at = EXCLUDED.updated_at,
                version = tasks.version + 1
            RETURNING id, (xmax = 0) AS inserted
        """)

        # Errors
        errors = await pg_conn.fetch("""
            SELECT external_id, error_msg
            FROM tmp_tasks_import WHERE NOT valid
        """)

    inserted = sum(1 for r in upsert_result if r["inserted"])
    updated = len(upsert_result) - inserted

    elapsed_ms = int((time.perf_counter() - start) * 1000)

    return BulkUpsertResponse(
        inserted=inserted,
        updated=updated,
        skipped=len(errors),
        errors=[{"external_id": e["external_id"], "error": e["error_msg"]} for e in errors],
        elapsed_ms=elapsed_ms,
    )

It needs a composite UNIQUE constraint (tenant_id, external_id):

# Migration: add unique
op.execute("""
    ALTER TABLE tasks
    ADD CONSTRAINT uq_tasks_tenant_external UNIQUE (tenant_id, external_id)
""")

And add external_id to the Task model. Create a migration that adds it + the constraint.

Include the router:

# app/main.py
from app.routers import tasks_bulk
app.include_router(tasks_bulk.router)

---

Zero-downtime migration: add priority NOT NULL

We'll execute the expand-contract pattern to add priority INTEGER NOT NULL DEFAULT 0 to tasks.

Setup: initial state

We assume priority already exists (we added it in the bulk endpoint), but we'll re-model the scenario: imagine it does NOT exist and you want to add it NOT NULL.

Drop it first (just for the exercise):

ALTER TABLE tasks DROP COLUMN priority;

Now we'll re-add it with expand-contract.

Migration 1: expand (add a nullable column with a default)

alembic revision -m "add priority column nullable with default"

# alembic/versions/XXX_add_priority_nullable.py
def upgrade() -> None:
    # Add nullable with default 0
    op.execute("""
        ALTER TABLE tasks
        ADD COLUMN priority INTEGER DEFAULT 0
    """)


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

This migration is safe: adding a nullable column with a default doesn't lock writes (PostgreSQL 11+).

Apply:

alembic upgrade head

The app keeps running without a restart. The column exists but the code doesn't use it yet (deploy 1).

Deploy 1: app that writes priority but defaults to 0

# app/models/task.py
class Task(Base):
    # ... other columns ...
    priority: Mapped[int] = mapped_column(Integer, default=0)

Restart the app:

docker-compose restart app

Now any task created has a priority. Existing rows have priority=0 (default).

Migration 2: backfill (not needed here because the default already covered it)

In this simplified case, default 0 already backfilled. If in another case you needed a per-row computation, it would be an UPDATE migration in batches.

Deploy 2: application reads priority normally

The queries start reading priority. Verify that all rows have a value:

SELECT COUNT(*) FROM tasks WHERE priority IS NULL;
-- Expected: 0 (all have default 0)

Migration 3: contract (make it NOT NULL)

alembic revision -m "make priority not null"

def upgrade() -> None:
    # Setting NOT NULL requires a full scan, but in PG 12+ it's a fast path
    # if the column has a default
    op.execute("""
        ALTER TABLE tasks ALTER COLUMN priority SET NOT NULL
    """)


def downgrade() -> None:
    op.execute("""
        ALTER TABLE tasks ALTER COLUMN priority DROP NOT NULL
    """)

Apply:

alembic upgrade head

This migration is the risky one. PostgreSQL validates that all rows have a value (it should be true because of the default). In PG 12+, if the column already has a default, the SCAN is a fast path.

Verify

SELECT column_name, is_nullable, column_default
FROM information_schema.columns
WHERE table_name = 'tasks' AND column_name = 'priority';
-- column_name | is_nullable | column_default
-- priority    | NO          | 0

---

Run it all live with wrk

The module's binary criterion: run the full migration with real traffic running and verify 0 errors.

wrk setup

post-task.lua:

-- post-task.lua: simulates creating tasks under load
wrk.method = "POST"
wrk.body = '{"project_id":"VALID-UUID-HERE","title":"Load test","status":"pending"}'
wrk.headers["Content-Type"] = "application/json"
wrk.headers["Authorization"] = "Bearer YOUR_TOKEN"

Migration script with load

#!/bin/bash
# scripts/run_migration_with_load.sh

set -e

echo "=== Zero-downtime migration with real traffic ==="
echo

# 1. Start wrk in the background
echo "Starting wrk for 600s..."
wrk -t 4 -c 50 -d 600s --latency \
    -s scripts/post-task.lua \
    http://localhost:8000/tasks > wrk_during_migration.log 2>&1 &
WRK_PID=$!

# 2. Wait 30s with traffic before starting
echo "Sleeping 30s..."
sleep 30

# 3. Migration 1: add nullable column
echo "Migration 1: ADD COLUMN priority INTEGER DEFAULT 0..."
docker-compose exec -T app alembic upgrade XXX_add_priority_nullable
sleep 60

# 4. Deploy 2: code that writes priority
echo "Deploy 2: app that writes priority..."
docker-compose restart app
sleep 60

# 5. Migration 3: SET NOT NULL
echo "Migration 3: SET NOT NULL..."
docker-compose exec -T app alembic upgrade head
sleep 60

# 6. Wait for wrk to finish
echo "Waiting for wrk to complete..."
wait $WRK_PID

# 7. Report
echo
echo "=== Results ==="
cat wrk_during_migration.log
echo
echo "=== Errors ==="
ERRORS=$(grep -c "Non-2xx or 3xx" wrk_during_migration.log || echo 0)
echo "HTTP errors: $ERRORS"

if [ "$ERRORS" = "0" ]; then
    echo "SUCCESS: zero downtime migration completed."
    exit 0
else
    echo "FAILURE: $ERRORS errors during migration."
    exit 1
fi

Run

chmod +x scripts/run_migration_with_load.sh
./scripts/run_migration_with_load.sh

Expected output:

=== Zero-downtime migration with real traffic ===

Starting wrk for 600s...
Sleeping 30s...
Migration 1: ADD COLUMN priority INTEGER DEFAULT 0...
Deploy 2: app that writes priority...
Migration 3: SET NOT NULL...
Waiting for wrk to complete...

=== Results ===
Running 10m test @ http://localhost:8000/tasks
  4 threads and 50 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    14.23ms    8.45ms  150.34ms   72.15%
    Req/Sec     1.2k      234.5     2.1k     68.34%
  Latency Distribution
     50%   12.45ms
     75%   18.12ms
     90%   24.67ms
     99%   45.89ms
  720,000 requests in 10.00m, 95.34MB read
Requests/sec:  1,200.00
Transfer/sec:  162.45KB

=== Errors ===
HTTP errors: 0
SUCCESS: zero downtime migration completed.

0 errors = success.

---

If errors show up

If HTTP errors: > 0, the deploy had momentary downtime. Investigate:

1. docker-compose restart takes too long. The app is slow to start — requests during that gap fail. Fix: use a reverse proxy (nginx) with an upstream healthcheck that removes the container from the pool before the restart.

2. The migration locks for too long. ALTER TABLE on a large table can take seconds. PostgreSQL scales well with ALTER TABLE ... ADD COLUMN nullable DEFAULT since PG 11 — fast path. If your PG is old, consider another approach.

3. PgBouncer pool exhausted during the restart. The old app's connections close abruptly. PgBouncer keeps its own but re-routes to the new app.

---

Pitfalls and common mistakes

1. ADD COLUMN ... NOT NULL DEFAULT 0 directly (without expand-contract).

On large tables, this operation locks the table rewrite — minutes of downtime. The safe way is:

1. ADD COLUMN nullable DEFAULT (fast).
2. Backfill if needed (not here because the default covered it).
3. SET NOT NULL (fast, PG 12+).

2. wrk with connections that close on restart.

-c 50 keeps 50 persistent connections. When the app restarts, those connections close. Wrk reconnects — during that gap, those requests fail. That's why the script has sleep 30 between operations; PgBouncer mitigates it.

3. wrk throwing so many requests per second it saturates.

If wrk generates more load than the app can handle, errors appear but not because of the migration. Calibrate -t 4 -c 50 so the app is at ~50% of its max capacity.

4. App crashing from a bug, not from the migration.

If your deploy 2 has a code bug (e.g., a typo in a query), the app can crash. That shows up as "errors" in wrk but it's not a migration problem. Test the deploy locally first.

5. A migration that really is blocking.

Some operations DO block writes:

• ADD COLUMN NOT NULL without a default (obvious).
• ADD CONSTRAINT when it validates existing data.
• CREATE INDEX (without CONCURRENTLY).

Check with EXPLAIN or pg_locks that it isn't holding an excessive lock.

6. docker-compose restart with downtime.

A restart takes 5-15 seconds. Without a reverse proxy, those 15s = 15s of errors. For a "real" zero-downtime migration: use Kubernetes with a rolling deployment, or nginx upstream with a healthcheck.

7. An undocumented RUNBOOK.

If the migration has a problem in production, someone has to diagnose it at 3am. Without an explicit RUNBOOK ("if X fails, run Y"), panic. Capsule 08 covers the RUNBOOK.

---

Summary and next step

What you have now:

• The POST /tasks/bulk endpoint with COPY + ON CONFLICT + automatic audit log.
• A zero-downtime migration executed live with wrk running.
• 0 errors in 600s — the binary criterion cleared.
• A reproducible script to run the migration with load.

Commit:

git add .
git commit -m "feat: bulk endpoint + zero-downtime migration with load testing"

In the next capsule we complete the integration tests: aggressive RLS isolation, full CRUD, optimistic locking, bulk endpoint, audit log. Tests with real Postgres (testcontainers), no mocks. And measured benchmarks to include in BENCHMARKS.md.

---

Resources

1. PostgreSQL — ALTER TABLE ... ADD COLUMN — official reference.
2. wrk — load testing tool.
3. PostgreSQL 11 — Fast ADD COLUMN — the change in PG 11.
4. Brandur — Postgres migrations — real patterns.
5. GitLab — Database migration guidelines — a real runbook.
6. Heroku — Pre-deploy checks — operational patterns.

---

Capsule 06 of 08 — Module 8 — SQL Patterns for Production APIs Guide