Practical Case: Bulk CSV Import Service

Scenario

Users upload a CSV (could be 5 rows, could be 500,000) to import records — products, contacts, transactions. You must validate every row, import the valid ones, report the invalid ones precisely (“row 4,213: email is malformed”), show progress while it runs, and survive a worker crash mid-import without double-importing or losing the job.

This is a very common senior take-home because the naive version — “read the file in the request, loop, insert” — breaks on every realistic input: a big file blows the request timeout and the worker’s memory, one bad row in the middle either aborts everything or gets silently skipped, and there’s no way to tell the user what happened. The senior design is async processing + streaming + per-row results + idempotent, resumable work.

Stack: FastAPI + Postgres + Celery + Redis + S3.

How would you design this?

Upload and processing are separate phases; the file goes to object storage, never through the worker’s memory all at once.

1. Upload — client uploads the CSV straight to S3 (pre-signed URL). The API creates an import_job row (status=pending) and enqueues a Celery task. Response is immediate with a job_id.
2. Process — a worker streams the file from S3 row by row, validates each row, batches the valid ones into the DB, records per-row outcomes, updates progress.
3. Report — client polls GET /imports/{job_id} for status/progress, and on completion downloads an error report (which rows failed and why).

[client] --pre-signed PUT--> [S3: raw.csv]
[client] --POST /imports--> [API: create import_job, enqueue]
                                       |
                                       v
                                  [Celery worker]
                              stream S3 -> validate -> batch INSERT
                                       |
                          [Postgres: records + import_rows + import_jobs]
[client] --GET /imports/{id}--> [API: progress + error report]

The job model

Three tables carry the whole design:

class ImportJob(Base):
    id            = Column(UUID, primary_key=True)
    s3_key        = Column(Text)
    status        = Column(Enum("pending","running","done","failed"))
    total_rows    = Column(Integer)         # filled in after first pass
    processed_rows = Column(Integer, default=0)
    ok_rows       = Column(Integer, default=0)
    failed_rows   = Column(Integer, default=0)
    last_row      = Column(Integer, default=0)   # for resume
    created_at    = Column(DateTime)

class ImportRow(Base):                       # one per *failed* row (or all rows)
    job_id        = Column(UUID, ForeignKey("import_job.id"))
    row_number    = Column(Integer)
    status        = Column(Enum("ok","error"))
    error         = Column(Text)

import_jobs is the unit the user polls. import_rows is the per-row audit trail that powers the error report. last_row is what makes the job resumable.

The worker — stream, don’t load

@celery_app.task(bind=True, max_retries=3)
def run_import(self, job_id: str):
    job = db.get(ImportJob, job_id)
    job.status = "running"; db.commit()

    obj = s3.get_object(Bucket=BUCKET, Key=job.s3_key)
    reader = csv.DictReader(io.TextIOWrapper(obj["Body"], encoding="utf-8"))

    batch, row_num = [], 0
    for row_num, raw in enumerate(reader, start=1):
        if row_num <= job.last_row:          # resume: skip already-done rows
            continue
        try:
            batch.append(validate(raw))      # pydantic model -> raises on bad row
        except ValidationError as e:
            record_error(job_id, row_num, str(e))

        if len(batch) >= 1000:
            flush(job, batch, row_num)       # bulk insert + checkpoint, one txn
            batch = []

    flush(job, batch, row_num)
    job.status = "done"; db.commit()

def flush(job, batch, row_num):
    with db.begin():                         # ONE transaction per batch
        db.execute(insert(Record), batch)    # bulk insert
        job.processed_rows += len(batch)
        job.ok_rows += len(batch)
        job.last_row = row_num               # checkpoint: resume point

Key points:

• Stream from S3 with csv.DictReader over the response body — constant memory regardless of file size. Never download_file then read() a 500MB CSV into RAM.
• Batch the inserts. One INSERT per row is thousands of round trips; batches of ~1,000 in a single multi-row insert is the difference between minutes and hours.
• Checkpoint last_row inside the batch transaction. The bulk insert and the last_row update commit together. If the worker crashes, last_row reflects exactly what’s durably in the DB — no more, no less.
• Resume on retry. Celery retries the task; the row_num <= job.last_row skip means it picks up where it left off instead of re-importing the first half. Combined with idempotency (below), a retry is safe.
• One bad row doesn’t abort the job. Validation failure → record the error, continue. The user gets a complete error report, not “import failed at row 4,213, the other 495,787 rows are in limbo.”

Idempotency — surviving retries and re-uploads

Two distinct duplication risks:

• Mid-job crash + retry — handled by last_row checkpointing: rows before the checkpoint are committed, rows after aren’t, resume skips the committed ones.
• Partial batch ambiguity — the resume point is transaction-aligned, so there’s no half-written batch. But to be defensive against the same logical record appearing twice (retry races, or the user uploading the same file twice), put a natural/business key on records with INSERT ... ON CONFLICT DO NOTHING (upsert). Then re-importing a row is a no-op, not a duplicate.

At-least-once execution + idempotent writes = effectively-once import. Same principle as the webhook service.

Progress and the error report

Progress is just reading the job row — processed_rows / total_rows:

@app.get("/imports/{job_id}")
async def status(job_id: UUID):
    job = await db.get(ImportJob, job_id)
    return {
        "status": job.status,
        "progress": job.processed_rows / (job.total_rows or 1),
        "ok": job.ok_rows, "failed": job.failed_rows,
        "error_report_url": presign(error_csv_key(job_id)) if job.failed_rows else None,
    }

The client polls this (or you push updates over WebSocket/SSE if the UX needs live progress — polling every few seconds is usually fine). The error report is import_rows WHERE status='error' rendered to a CSV and dropped in S3; the user downloads “here are your 213 bad rows and why.”

total_rows needs a count — either a cheap first pass over the file to count lines before the real pass, or accept “progress unknown until done” for huge files. A first pass over a streamed file is fast (no parsing, just newline counting).

On AWS — what runs where

For genuinely huge files, chunk the job: one orchestrator task splits the file into N byte-range chunks, fans out N worker tasks (Celery group/chord), each imports its range; the chord callback marks the job done. This parallelizes the import and bounds any single task’s runtime.

What can go wrong

• Loading the whole file into memory — a 500MB CSV OOMs the worker. Stream it.
• Row-at-a-time inserts — thousands of round trips; batch them.
• One transaction for the whole file — a crash at row 499,000 rolls back everything; and the lock/WAL pressure is brutal. One transaction per batch.
• Aborting on the first bad row — users want a full error report, not a partial import they can’t reason about. Validate-and-continue.
• No checkpoint — a retry re-imports from row 1, double-importing everything before the crash. Checkpoint last_row transactionally.
• No idempotency key on records — retries and re-uploads create duplicates. Natural key + ON CONFLICT DO NOTHING.
• Encoding assumptions — real CSVs are UTF-8 with BOM, Latin-1, CP1251, etc. Detect or let the user specify; fail the row clearly, don’t crash the job.
• Synchronous upload + parse in the request — request timeout on big files, no progress, no retry. Async from the start.

Interview angle

1. “The file is 2GB. What breaks, and what do you change?” — Loading it into memory OOMs the worker, and a single task may exceed the time limit. Stream from S3 (constant memory), keep batch inserts, and chunk the job into parallel byte-range sub-tasks (Celery chord) so no single task runs unbounded.
2. “The worker crashes at row 300,000 of 500,000. What’s the state, and what happens on retry?” — Rows up to the last committed batch are durably in the DB and last_row points at exactly that row (they commit together). On retry the task streams the file again but skips row_num <= last_row, resuming cleanly. Idempotent inserts make even an imprecise resume safe.
3. “How do you report errors precisely?” — Per-row validation; every failure writes an import_rows record with the row number and reason. On completion, render the error rows to a CSV in S3 and hand the user a download link. The job never aborts on a bad row.
4. “How do you show progress?” — The worker increments processed_rows in each batch transaction; the client polls GET /imports/{id} which returns processed/total. Live updates can go over WebSocket/SSE, but polling every few seconds is usually enough.
5. “User uploads the same file twice by accident — what happens?” — Each upload is a separate import_job, but the records table has a natural-key unique constraint with ON CONFLICT DO NOTHING, so the second import inserts nothing new. The job completes “successfully” with 0 new rows — visible and harmless, not a silent duplication.
6. “Why not validate everything first, then import?” — Two full passes double the I/O, and for a streamed file you can’t cheaply “go back.” Validate-and-import in one streaming pass; per-row errors are recorded inline, valid rows are batched as you go.

Cross-links:

• LLM file pipeline (object storage + worker pattern): 01_llm_file_processing_pipeline.md
• Webhook service (idempotency, at-least-once): 02_webhook_ingestion_service.md
• Bulk insert / DB performance: ../backend/08_databases/sql/
• Celery vs asyncio: ../system_design/03_async_patterns/01_async_io_and_background_work.md