Module 7: Bulk Operations

Mini-project: `POST /tasks/bulk` with benchmarks

Next →

You close the module with a complete TaskFlow endpoint that integrates everything you learned. POST /tasks/bulk accepts up to 50,000 tasks, inserts them using COPY + ON CONFLICT with a temp table, validates in SQL, returns granular stats, and measures the total time. The deliverable is production-ready code with documented benchmarks that you link in your portfolio.

This endpoint is exactly the kind of feature that separates a "toy API" from an "enterprise-ready API". Any developer POSTs a single task; few do a bulk POST with 50k tasks that finishes in under a second, with granular error reporting and perfect idempotency.

Endpoint specification

Request

POST /tasks/bulk HTTP/1.1
Content-Type: application/json

[
  {"external_id": "ext-1", "title": "Task 1", "status": "pending", "priority": 5},
  {"external_id": "ext-2", "title": "Task 2", "status": "completed", "priority": 3},
  ...
]

Response (success)

{
  "imported": 8523,
  "updated": 1477,
  "skipped": 0,
  "errors": [],
  "elapsed_ms": 1450,
  "throughput_per_sec": 6896
}

Response (with partial errors)

{
  "imported": 8500,
  "updated": 1450,
  "skipped": 50,
  "errors": [
    {"external_id": "ext-3", "error": "invalid status"},
    {"external_id": "ext-7", "error": "priority out of range"},
    ...
  ],
  "elapsed_ms": 1500,
  "throughput_per_sec": 6633
}

Limits

  • Max 50,000 tasks per request.
  • Max body size 10MB.
  • Validation rules:
    • external_id: required, max 100 chars.
    • title: required, 1-200 chars.
    • status: must be in ['pending', 'in_progress', 'completed', 'archived'].
    • priority: required, integer 1-5.

Complete implementation

Schema

# app/schemas.py
from pydantic import BaseModel, Field, ConfigDict


class TaskBulkItem(BaseModel):
    model_config = ConfigDict(extra='forbid')  # strict request body

    external_id: str = Field(min_length=1, max_length=100)
    title: str = Field(min_length=1, max_length=200)
    status: str  # validation per row in SQL
    priority: int = Field(ge=1, le=5)


class BulkErrorItem(BaseModel):
    external_id: str
    error: str


class BulkUpsertResponse(BaseModel):
    imported: int
    updated: int
    skipped: int
    errors: list[BulkErrorItem]
    elapsed_ms: int
    throughput_per_sec: int

Model

# app/models.py
from datetime import datetime, timezone
from sqlalchemy import String, Integer, 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)
    external_id: Mapped[str] = mapped_column(String(100), unique=True)
    title: Mapped[str] = mapped_column(String(200))
    status: Mapped[str] = mapped_column(String(50))
    priority: Mapped[int] = mapped_column(Integer)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
        onupdate=lambda: datetime.now(timezone.utc),
    )

Endpoint

# app/routers/bulk.py
import time
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.schemas import TaskBulkItem, BulkUpsertResponse, BulkErrorItem


router = APIRouter()

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


@router.post("/tasks/bulk", response_model=BulkUpsertResponse)
async def bulk_upsert_tasks(
    tasks: list[TaskBulkItem],
    db: AsyncSession = Depends(get_db),
) -> BulkUpsertResponse:
    """Bulk upsert of tasks.

    - Up to 50k tasks per request.
    - SQL-level validation (fast).
    - Idempotent via ON CONFLICT (external_id).
    - Granular reports with per-row errors.
    """
    # Limit check
    if len(tasks) == 0:
        raise HTTPException(400, "Empty payload")
    if len(tasks) > MAX_TASKS:
        raise HTTPException(
            status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail=f"Max {MAX_TASKS} tasks per request"
        )

    start = time.perf_counter()

    # Convert Pydantic models to tuples
    now = datetime.now(timezone.utc)
    records = [
        (t.external_id, t.title, t.status, t.priority, now, now)
        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():
        # 1. Create temp table
        await pg_conn.execute("""
            CREATE TEMP TABLE tmp_tasks_import (
                external_id TEXT,
                title TEXT,
                status TEXT,
                priority INT,
                created_at TIMESTAMPTZ,
                updated_at TIMESTAMPTZ,
                valid BOOL DEFAULT TRUE,
                error_msg TEXT
            ) ON COMMIT DROP
        """)

        # 2. COPY records into temp
        await pg_conn.copy_records_to_table(
            "tmp_tasks_import",
            records=records,
            columns=["external_id", "title", "status", "priority",
                     "created_at", "updated_at"],
        )

        # 3. SQL-level validation
        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)})
        """)

        # 4. Detect duplicates within the payload (keep the last one)
        await pg_conn.execute("""
            DELETE FROM tmp_tasks_import a
            USING tmp_tasks_import b
            WHERE a.external_id = b.external_id
              AND a.ctid < b.ctid
              AND b.valid = TRUE
        """)

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

        # 6. Collect errors
        errors = await pg_conn.fetch("""
            SELECT external_id, error_msg
            FROM tmp_tasks_import
            WHERE valid = FALSE
        """)

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

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

    total_processed = inserted + updated + skipped
    throughput = int(total_processed / elapsed_seconds) if elapsed_seconds > 0 else 0

    return BulkUpsertResponse(
        imported=inserted,
        updated=updated,
        skipped=skipped,
        errors=[
            BulkErrorItem(external_id=e["external_id"], error=e["error_msg"])
            for e in errors
        ],
        elapsed_ms=elapsed_ms,
        throughput_per_sec=throughput,
    )

Tests

Test 1: happy path

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


@pytest.mark.asyncio
async def test_bulk_insert_new_tasks(client: AsyncClient):
    payload = [
        {"external_id": "ext-1", "title": "T1", "status": "pending", "priority": 5},
        {"external_id": "ext-2", "title": "T2", "status": "completed", "priority": 3},
    ]

    response = await client.post("/tasks/bulk", json=payload)
    assert response.status_code == 200

    body = response.json()
    assert body["imported"] == 2
    assert body["updated"] == 0
    assert body["skipped"] == 0
    assert body["errors"] == []
    assert body["elapsed_ms"] >= 0

Test 2: idempotency

@pytest.mark.asyncio
async def test_bulk_upsert_is_idempotent(client: AsyncClient):
    payload = [{"external_id": "ext-idem", "title": "T", "status": "pending", "priority": 5}]

    # First call
    r1 = await client.post("/tasks/bulk", json=payload)
    assert r1.json()["imported"] == 1

    # Second call - should update
    r2 = await client.post("/tasks/bulk", json=payload)
    assert r2.json()["updated"] == 1
    assert r2.json()["imported"] == 0

Test 3: partial success with invalid

@pytest.mark.asyncio
async def test_bulk_partial_success_with_invalid(client: AsyncClient):
    payload = [
        {"external_id": "valid-1", "title": "T1", "status": "pending", "priority": 5},
        {"external_id": "invalid-1", "title": "T2", "status": "BAD_STATUS", "priority": 3},
        {"external_id": "valid-2", "title": "T3", "status": "completed", "priority": 1},
    ]

    response = await client.post("/tasks/bulk", json=payload)
    body = response.json()

    assert body["imported"] == 2
    assert body["skipped"] == 1
    assert len(body["errors"]) == 1
    assert body["errors"][0]["external_id"] == "invalid-1"

Test 4: duplicate within the same payload

@pytest.mark.asyncio
async def test_bulk_duplicates_in_payload(client: AsyncClient):
    payload = [
        {"external_id": "dup-1", "title": "First", "status": "pending", "priority": 5},
        {"external_id": "dup-1", "title": "Last", "status": "pending", "priority": 5},
    ]

    response = await client.post("/tasks/bulk", json=payload)
    body = response.json()

    assert body["imported"] == 1  # only 1 final row

    # Verify the last one won (after the DELETE in temp)
    # ... fetch task by external_id, verify title == "Last"

Test 5: limits

@pytest.mark.asyncio
async def test_bulk_too_many_tasks(client: AsyncClient):
    payload = [
        {"external_id": f"ext-{i}", "title": f"T{i}", "status": "pending", "priority": 1}
        for i in range(50_001)
    ]

    response = await client.post("/tasks/bulk", json=payload)
    assert response.status_code == 413


@pytest.mark.asyncio
async def test_bulk_empty_payload(client: AsyncClient):
    response = await client.post("/tasks/bulk", json=[])
    assert response.status_code == 400

Test 6: Pydantic validation (request body)

@pytest.mark.asyncio
async def test_bulk_invalid_payload_pydantic(client: AsyncClient):
    payload = [
        {"external_id": "ext-1", "title": "T", "status": "pending"},  # missing priority
    ]
    response = await client.post("/tasks/bulk", json=payload)
    assert response.status_code == 422  # Pydantic validation error

Benchmarks

Benchmark setup

# benchmark.py
import asyncio
import time
import httpx


async def benchmark(N: int):
    payload = [
        {
            "external_id": f"ext-{i}",
            "title": f"Task {i}",
            "status": "pending",
            "priority": (i % 5) + 1,
        }
        for i in range(N)
    ]

    async with httpx.AsyncClient(base_url="http://localhost:8000", timeout=30) as client:
        start = time.perf_counter()
        response = await client.post("/tasks/bulk", json=payload)
        elapsed = time.perf_counter() - start

        body = response.json()
        print(f"\nN={N}")
        print(f"  Server elapsed_ms: {body['elapsed_ms']}ms")
        print(f"  Total roundtrip: {elapsed*1000:.0f}ms")
        print(f"  Imported: {body['imported']}, Updated: {body['updated']}")
        print(f"  Throughput: {body['throughput_per_sec']:,}/s")


async def main():
    # Cleanup
    # ... truncate table

    for N in [100, 1_000, 10_000, 50_000]:
        await benchmark(N)


asyncio.run(main())

Typical results

N=100
  Server elapsed_ms: 8ms
  Total roundtrip: 23ms
  Imported: 100, Updated: 0
  Throughput: 12,500/s

N=1,000
  Server elapsed_ms: 32ms
  Total roundtrip: 89ms
  Imported: 1,000, Updated: 0
  Throughput: 31,250/s

N=10,000
  Server elapsed_ms: 145ms
  Total roundtrip: 412ms
  Imported: 10,000, Updated: 0
  Throughput: 68,966/s

N=50,000
  Server elapsed_ms: 680ms
  Total roundtrip: 1,890ms
  Imported: 50,000, Updated: 0
  Throughput: 73,529/s

Sub-linear throughput but acceptable. Server processing is <1s for 50k. Total roundtrip ~2s (includes JSON serialization, network, etc.).

Re-import test (idempotency)

async def benchmark_reimport(N: int):
    # First time: insert
    await benchmark(N)

    # Second time: update all
    await benchmark(N)


# Results:
# First (inserts): 680ms server-side
# Second (updates): 720ms server-side
# Third (identical updates): 700ms server-side

Updates are slightly more expensive than inserts (they check the constraint, run an UPDATE instead of a plain INSERT). Acceptable.

The BENCHMARKS.md for your portfolio

# TaskFlow — Bulk Operations Endpoint

API endpoint that imports up to 50,000 tasks in under 1 second
of server-side processing.

## Setup

- PostgreSQL 16
- FastAPI 0.110+
- SQLAlchemy 2.0+ (async) + asyncpg 0.29+
- Hardware: M2 Pro, 16GB RAM, NVMe SSD

## Performance

| N tasks | elapsed_ms (server) | throughput |
|---------|---------------------|------------|
| 100     | 8ms                 | 12.5k/s    |
| 1,000   | 32ms                | 31k/s      |
| 10,000  | 145ms               | 69k/s      |
| 50,000  | 680ms               | 74k/s      |

## Implemented pattern

1. **COPY into a temp table** (`copy_records_to_table` with asyncpg).
2. **SQL-level validation** (UPDATE temp SET valid = FALSE WHERE ...).
3. **DELETE duplicates** within the payload.
4. **INSERT...SELECT...ON CONFLICT** from temp into real.
5. **Granular stats** via `(xmax = 0) AS inserted`.

## Features

- ✅ Idempotency via `ON CONFLICT (external_id)`.
- ✅ Atomic validation with SQL.
- ✅ Partial success (invalid rows reported, valid ones imported).
- ✅ Safety limits (max 50k, max body 10MB).
- ✅ Useful stats in the response.
- ✅ Automated tests (pytest + httpx).

## Reproduce

```bash
docker-compose up -d
alembic upgrade head
uvicorn app.main:app
pytest tests/
python benchmark.py

Learnings

  1. COPY + temp + INSERT...ON CONFLICT is ~10x faster than pg_insert.on_conflict_do_update for 50k+ rows.
  2. SQL-level validation is ~50x faster than iterating in Python.
  3. DELETE of duplicates with ctid inside temp is elegant.
  4. (xmax = 0) AS inserted is the trick to tell INSERT vs UPDATE apart.

Next optimizations

  • Streaming endpoint for >50k rows (with SSE for progress).
  • Background job for imports >>1M rows.
  • Prometheus metrics for throughput, error rate.
  • Per-user rate limiting.

---

## Module wrap-up

What you learned across the 8 capsules:

1. **Capsule 01:** Introduction and a decision matrix by size.
2. **Capsule 02:** The 4 approaches with benchmarks (47s → 0.8s).
3. **Capsule 03:** COPY in depth with asyncpg.
4. **Capsule 04:** `bulk_insert_mappings` and its limitations.
5. **Capsule 05:** Atomic upserts with `ON CONFLICT`.
6. **Capsule 06:** Bulk upserts with a temp table (canonical).
7. **Capsule 07:** Error handling + streaming.
8. **Capsule 08:** The integrating mini-project (this one).

Module wrap-up:

- The canonical pattern for large imports internalized.
- A production-ready endpoint on public GitHub.
- Documented performance metrics.
- Automated tests covering edge cases.

---

## Getting started in the next module

**Module 8** consolidates the 7 patterns into a complete TaskFlow — the guide's final integrating project. A multi-tenant SaaS API that applies:

- Cursor pagination (module 1).
- Soft deletes (module 2).
- Audit logs (module 3).
- Multi-tenancy with RLS (module 4).
- Zero-downtime migrations (module 5).
- Optimistic locking + schema versioning (module 6).
- Bulk operations with COPY + ON CONFLICT (module 7).

It's the guide's "magnum opus" — when you finish it, your portfolio will have an API that demonstrates integrated mastery of all the production-ready patterns.

---

## Resources

1. [GitHub README templates](https://github.com/othneildrew/Best-README-Template) — for your portfolio.
2. [PostgreSQL Docs — `COPY`](https://www.postgresql.org/docs/current/sql-copy.html) — official reference.
3. [asyncpg — `copy_records_to_table`](https://magicstack.github.io/asyncpg/current/api/index.html#asyncpg.Connection.copy_records_to_table) — reference.
4. [FastAPI — Bulk operations patterns](https://fastapi.tiangolo.com/) — ecosystem.
5. [pytest-asyncio](https://pytest-asyncio.readthedocs.io/) — async testing.
6. [Brandur Leach — Postgres bulk insert](https://brandur.org/postgres-queries) — deep dive.
7. [Heap — Idempotent ETL](https://www.heap.io/blog/idempotent-etl-jobs) — real patterns at scale.

---

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

*End of module 7. Continue with module 8 (TaskFlow Final Project).*
← PreviousRobust error handling + streaming for gigantic datasetsNext →Module 8: Final Project — TaskFlow API