Module 7: Bulk Operations

Robust error handling + streaming for gigantic datasets

Next →

The patterns from capsules 02-06 cover the "happy path" case — everything works, valid data, the dataset fits in memory. In real production there are edge cases: the CSV file has a row with weird encoding, the client sends 50M rows that don't fit in memory, the connection drops in the middle of an import. This capsule covers the patterns to handle all those cases.

Three topics in this capsule: transactional error handling (fatal vs recoverable, partial success), streaming to avoid OOM, and progress reporting during long imports.

Types of errors in bulk operations

Fatal errors

  • Unhandleable constraint violation: an INSERT violates NOT NULL on a required column with no default.
  • Foreign key violation: you insert a task with an owner_id that doesn't exist in users.
  • Type mismatch: you pass a string to an INTEGER column.
  • Lost connection: the network is cut during a COPY.
  • Disk full: PostgreSQL runs out of space.

These abort the whole transaction. The batch stays as it was (automatic rollback).

Recoverable errors

  • Individual rows with invalid data: status isn't a valid enum, priority out of range.
  • Expected duplicates: ON CONFLICT handles it.
  • Data to transform: trim spaces, lowercase emails, etc.

These are recoverable with the right strategies.

Strategy 1: validate-everything-then-import

Validate in Python before touching the DB. If everything passes, COPY+INSERT. If something fails, return the errors without touching the DB.

async def validate_then_import(raw_data: list[dict], conn):
    valid_records = []
    errors = []

    for i, row in enumerate(raw_data):
        try:
            validated = TaskCreate(**row)  # Pydantic
            valid_records.append((
                validated.external_id,
                validated.title,
                validated.status,
                validated.priority,
            ))
        except ValidationError as e:
            errors.append({"index": i, "row": row, "errors": e.errors()})

    # If there's ANY error, fail-all
    if errors:
        return {
            "imported": 0,
            "errors": errors,
            "skipped_all_due_to_errors": True,
        }

    # If there are no errors, COPY+INSERT
    await copy_to_real_table(valid_records, conn)
    return {"imported": len(valid_records), "errors": []}

Pros:

  • Clear atomicity: all or nothing.
  • The client knows the data is good before any persist.
  • No weird partial states.

Cons:

  • If one invalid row out of 50k, fail-all can be frustrating.
  • Validation in Python is slower than in SQL.

Strategy 2: validate-as-import (partial success)

Allow partial success. Valid rows get imported, invalid ones are skipped with a warning.

async def validate_as_import(raw_data: list[dict], conn):
    valid_records = []
    errors = []

    for i, row in enumerate(raw_data):
        try:
            validated = TaskCreate(**row)
            valid_records.append((
                validated.external_id,
                validated.title,
                validated.status,
                validated.priority,
            ))
        except ValidationError as e:
            errors.append({"index": i, "row": row, "errors": e.errors()})

    # Import only the valid rows
    if valid_records:
        await copy_to_real_table(valid_records, conn)

    return {
        "imported": len(valid_records),
        "skipped": len(errors),
        "errors": errors,
    }

Pros:

  • Tolerance for partial errors.
  • The user has flexibility.

Cons:

  • "Imported some, errored some" is a complicated state for clients.
  • The client has to re-process errors manually.

Strategy 3: SQL-level validation with a temp table

Vectorized validation in SQL (faster for large datasets). The capsule 06 pattern with UPDATE temp SET valid = FALSE.

Pros:

  • Speed — SQL is orders of magnitude faster.
  • Validation rules stay versioned with migrations.

Cons:

  • Business rules end up in SQL (mixed with the persistence layer).
  • Harder to test.

When to use it:

  • Dataset >100k rows — Python validation is the bottleneck.
  • Validation rules expressible in SQL (format, ranges, enum values, foreign keys).

When NOT to use it:

  • Complex rules that require Python logic (complex regexes, business decisions).
  • When the errors need to be very descriptive.

Strategy 4: chunked processing

For gigantic datasets, process in chunks of N rows:

async def chunked_import(records, conn, chunk_size=10_000):
    total = len(records) if hasattr(records, "__len__") else "unknown"
    imported = 0
    errors = []

    chunks = (records[i:i+chunk_size] for i in range(0, len(records), chunk_size))

    for chunk_idx, chunk in enumerate(chunks):
        try:
            async with conn.transaction():
                await import_chunk(chunk, conn)
                imported += len(chunk)
            print(f"Chunk {chunk_idx}: imported {len(chunk)} rows. Total: {imported}/{total}")
        except Exception as e:
            errors.append({"chunk": chunk_idx, "error": str(e)})
            # Continue with the next chunk (chunked = recovery)

    return {"imported": imported, "errors": errors}

When to use it:

  • Datasets >1M rows.
  • Long-running imports (several minutes).
  • You want progress reporting to the client.

Pros:

  • Each chunk is an independent transaction — one failing doesn't affect the others.
  • Progress is reportable.
  • Memory bounded by chunk_size.

Cons:

  • Not globally atomic — if you fail on chunk 50 of 100, you have 49 chunks imported.
  • More code.

Streaming for gigantic datasets

To avoid OOM with datasets of millions, use generators.

Streaming from the HTTP request body

from fastapi import Request
import csv
import io


@router.post("/tasks/bulk-stream")
async def bulk_stream(request: Request, db: AsyncSession = Depends(get_db)):
    raw_conn = await db.connection()
    asyncpg_conn = await raw_conn.get_raw_connection()
    pg_conn = asyncpg_conn.driver_connection

    async def stream_records():
        """Generator that yields records from the request body."""
        buffer = b""
        async for chunk in request.stream():
            buffer += chunk
            # Process line by line
            while b"\n" in buffer:
                line, buffer = buffer.split(b"\n", 1)
                if line.strip():
                    fields = line.decode("utf-8").split(",")
                    yield (fields[0], fields[1], fields[2], int(fields[3]))

        # Final line
        if buffer.strip():
            fields = buffer.decode("utf-8").split(",")
            yield (fields[0], fields[1], fields[2], int(fields[3]))

    async with pg_conn.transaction():
        await pg_conn.execute("CREATE TEMP TABLE tmp_import (...) ON COMMIT DROP")

        await pg_conn.copy_records_to_table(
            "tmp_import",
            records=stream_records(),  # generator
            columns=["external_id", "title", "status", "priority"],
        )

        await pg_conn.execute("""
            INSERT INTO tasks SELECT * FROM tmp_import
            ON CONFLICT (external_id) DO UPDATE SET ...
        """)

    return {"status": "streamed"}

The client sends a large request. The server processes it in streaming. Memory kept under control.

Streaming from a file on disk

async def import_from_huge_file(file_path: str, conn):
    def parse_csv():
        with open(file_path) as f:
            reader = csv.DictReader(f)
            for row in reader:
                yield (
                    row["external_id"],
                    row["title"],
                    row["status"],
                    int(row["priority"]),
                )

    async with conn.transaction():
        await conn.execute("CREATE TEMP TABLE tmp_import (...) ON COMMIT DROP")
        await conn.copy_records_to_table("tmp_import", records=parse_csv(), columns=[...])
        await conn.execute("INSERT INTO tasks SELECT * FROM tmp_import ...")

Memory benchmark

ApproachRAM used for 1M rowsRAM for 10M rows
List in memory~200 MB~2 GB (OOM on many servers)
Generator streaming~10 MB~10 MB (constant)

Streaming is the difference between "your app crashes" and "your app keeps running".

Progress reporting

For long imports (minutes), the client wants to know the progress. There are three patterns:

1. Sync with WebSocket

@app.websocket("/ws/import")
async def import_with_progress(websocket: WebSocket):
    await websocket.accept()

    # Receive data over WebSocket
    data = await websocket.receive_json()
    records = data["records"]

    chunk_size = 1000
    total = len(records)

    for i, chunk in enumerate(chunks(records, chunk_size)):
        await import_chunk(chunk)
        progress = (i + 1) * chunk_size / total * 100
        await websocket.send_json({
            "progress": progress,
            "imported": (i + 1) * chunk_size,
            "total": total,
        })

    await websocket.send_json({"status": "complete"})

2. Background job + polling endpoint

@router.post("/tasks/bulk-async")
async def bulk_async(records: list[TaskBulk]):
    job = ImportJob.create(records=records)
    asyncio.create_task(process_in_background(job))
    return {"job_id": job.id}


@router.get("/tasks/bulk-async/{job_id}/status")
async def get_job_status(job_id: str):
    job = await ImportJob.get(job_id)
    return {
        "status": job.status,
        "imported": job.imported,
        "total": job.total,
        "progress": job.progress_percent,
        "errors": job.errors,
    }

The client makes the initial POST, then polls status until status == "complete".

3. Server-Sent Events (SSE)

from fastapi.responses import StreamingResponse


@router.post("/tasks/bulk-sse")
async def bulk_sse(records: list[TaskBulk]):
    async def event_stream():
        chunk_size = 1000
        total = len(records)

        for i, chunk in enumerate(chunks(records, chunk_size)):
            await import_chunk(chunk)
            progress = (i + 1) * chunk_size / total * 100
            yield f"data: {json.dumps({'progress': progress})}\n\n"

        yield "data: {\"status\": \"complete\"}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

The client uses the EventSource API (native in browsers).

Recovery patterns

If a long import fails halfway through, how do you recover?

Pattern 1: idempotent imports

If the import is idempotent (via ON CONFLICT), simply re-run it. Rows already imported get "updated" silently, missing rows get inserted.

Pattern 2: checkpoint files

For very long imports, keep a checkpoint of the progress:

async def import_with_checkpoint(records, checkpoint_file):
    last_processed = 0
    if os.path.exists(checkpoint_file):
        with open(checkpoint_file) as f:
            last_processed = int(f.read())

    for i, chunk in enumerate(chunks(records, 1000)):
        if i * 1000 < last_processed:
            continue  # already processed in a previous run

        await import_chunk(chunk)

        # Update checkpoint
        with open(checkpoint_file, "w") as f:
            f.write(str((i + 1) * 1000))

Pattern 3: idempotency key per chunk

async def import_chunk_idempotent(chunk_id, chunk_data, conn):
    # Check whether this chunk was already processed
    existing = await conn.fetchval(
        "SELECT 1 FROM import_chunks WHERE chunk_id = $1",
        chunk_id
    )
    if existing:
        return  # already processed, skip

    async with conn.transaction():
        await import_chunk(chunk_data, conn)
        await conn.execute(
            "INSERT INTO import_chunks (chunk_id) VALUES ($1)",
            chunk_id
        )

The import_chunks table tracks which chunks already passed. Re-running the import skips the processed ones.

Pitfalls and common mistakes

1. request.stream() consumed twice.

FastAPI consumes the body when it parses Pydantic models. If you want streaming, declare the endpoint without a Pydantic body and use request.stream() directly.

2. Generator that opens a file and closes it before consuming it.

def parse():
    with open("data.csv") as f:  # f closes when the generator exits
        reader = csv.reader(f)
        for row in reader:
            yield row

# The generator is consumed LATER. f is already closed → error.

Better:

def parse():
    f = open("data.csv")
    try:
        reader = csv.reader(f)
        for row in reader:
            yield row
    finally:
        f.close()

Or the contextlib pattern:

@contextlib.contextmanager
def csv_records(file_path):
    with open(file_path) as f:
        yield csv.reader(f)

3. Background tasks without persistent state.

If your app crashes during a background import, you lose the state. For important imports, persist the job state in the DB (not just in memory).

4. Streaming without periodic commits.

# ❌ The whole transaction in one
async with conn.transaction():
    for record in stream:
        await insert(record)
# If the stream has 10M records, the transaction lasts hours, locks held that whole time

For VERY long streams, consider periodic commits (every 100k records).

5. WebSocket that loses the connection halfway.

If the client disconnects, your import keeps running (server side) but the client doesn't receive progress updates. Consider a background job with polling as a more robust alternative.

6. SSE without reconnection logic.

The browser's EventSource auto-reconnects, but the server has to be able to resume from where it left off (it sends Last-Event-ID in the headers). Implement it correctly.

7. Reporting too granular.

Sending a progress update for each record is overhead. Every 1% or every 1k records is reasonable.

8. Not validating the size before processing.

if len(records) > 10_000_000:
    raise HTTPException(413, "Too large, use bulk upload endpoint")

Without checks, a client can saturate servers with an enormous request.

Exercise: complete pipeline with error handling

Setup: a Task model with external_id UNIQUE.

Step 1: implement the endpoint with strategy 3 (SQL-level validation + temp).

@router.post("/tasks/bulk-robust")
async def bulk_robust(records: list[TaskBulk], db = Depends(get_db)):
    # Validate size
    if len(records) > 100_000:
        raise HTTPException(413)

    # COPY into temp + validate in SQL
    # ... implement

Step 2: test with a small dataset that has known errors.

test_data = [
    {"external_id": "valid-1", "title": "T1", "status": "pending", "priority": 5},
    {"external_id": "valid-2", "title": "T2", "status": "completed", "priority": 3},
    {"external_id": "invalid-1", "title": "T3", "status": "BAD_STATUS", "priority": 5},
    {"external_id": "invalid-2", "title": "T4", "status": "pending", "priority": 99},  # priority > 5
    {"external_id": "valid-1", "title": "T1 dup", "status": "completed", "priority": 5},  # duplicate
]

Expect:

  • inserted: 2 (valid-1, valid-2)
  • updated: 1 (valid-1 with the value from the duplicate)
  • skipped: 2 (invalid-1, invalid-2)

Step 3: implement the streaming endpoint.

@router.post("/tasks/bulk-stream")
async def bulk_stream(request: Request, db = Depends(get_db)):
    # Stream from the request body
    # ... implement

Test with a large file:

# Generate a large file
python -c "
import csv
with open('huge.csv', 'w') as f:
    w = csv.writer(f)
    for i in range(1_000_000):
        w.writerow([f'ext-{i}', f'T{i}', 'pending', i % 5 + 1])
"

# Stream upload
curl -X POST http://localhost:8000/tasks/bulk-stream \
  --data-binary @huge.csv \
  -H "Content-Type: text/csv"

Check the server's memory during the upload (it should be <100MB).

Step 4: implement progress reporting with SSE.

(Code in the "Server-Sent Events" section above)

Test from a JavaScript client:

const source = new EventSource("/tasks/bulk-sse");
source.onmessage = (e) => {
  const data = JSON.parse(e.data);
  console.log("Progress:", data);
};

Step 5: test recovery: kill the server halfway through the import, restart, verify that the import can be re-run without errors (idempotent).

See discussion

Step 1 — robust:

Combining SQL validation + temp + ON CONFLICT gives you the complete pattern. High speed, clear error handling, idempotent.

Step 2 — partial success:

The test confirms the pattern handles every case:

  • Validation done in SQL (vectorized).
  • Duplicates handled by ON CONFLICT.
  • Invalids reported but no fail-all.

Step 3 — streaming:

Stable memory during a stream of 1M rows. Server CPU is used but not memory.

Step 4 — progress:

The client receives regular updates. Significantly better UX for long imports.

Step 5 — recovery:

Idempotency via ON CONFLICT lets you re-run without problems. Rows already imported get "updated" (to the same values), new ones get inserted. Never duplicates.

Key takeaways:

  1. The strategy choice depends on the case: validate-all-then-import for financial data; partial-success for tolerant imports; chunked for very large ones.
  2. Streaming is mandatory for datasets >> available memory.
  3. Progress reporting improves UX for long operations.
  4. Idempotency with ON CONFLICT simplifies recovery dramatically.

Summary and next step

What you learned:

  • Fatal vs recoverable errors: constraint violations vs individual invalid rows.
  • 4 strategies: validate-then-import (atomic), validate-as-import (partial), SQL-level (fast), chunked (recovery).
  • Streaming: generators to avoid OOM on gigantic datasets.
  • Progress reporting: WebSocket, polling, SSE — pick based on the case.
  • Recovery patterns: idempotency (best), checkpoint files, idempotency keys per chunk.
  • Pitfalls: request.stream() consumed twice, generators with a closed file, background tasks without persistent state.

Before moving on, you should be able to:

  • Choose the right strategy based on size and requirements.
  • Implement streaming for datasets that don't fit in memory.
  • Design recovery patterns for long imports.
  • Report progress to the client appropriately.

In the next capsule you close the module with the mini-project: a complete POST /tasks/bulk endpoint for TaskFlow with documented benchmarks. You'll integrate everything you learned — COPY + ON CONFLICT + temp table + streaming + progress reporting + tests — into a production-ready implementation that you link in your portfolio.

Resources

  1. FastAPI — Streaming Response — streaming output reference.
  2. FastAPI — Request body streaming — request streaming.
  3. MDN — Server-Sent Events — SSE reference.
  4. PostgreSQL — Setting transaction timeouts — statement_timeout, etc.
  5. Heap — Idempotent ETL — real patterns.
  6. asyncpg — Streaming patterns — reference.
  7. Python contextlib — for resource management in generators.

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

← PreviousBulk upserts with a temp table: the canonical patternNext →Mini-project: `POST /tasks/bulk` with benchmarks