Streaming CSV parse and generate patterns for large file support across Next.js frontend and FastAPI backend. Covers row-by-row streaming, Zod row validation, Pydantic export models, Australian locale formatting (DD/MM/YYYY dates, AUD currency), and memory-safe chunked processing.
Streaming CSV parse and generate patterns for large file support across Next.js frontend and FastAPI backend. Covers row-by-row streaming, Zod row validation, Pydantic export models, Australian locale formatting (DD/MM/YYYY dates, AUD currency), and memory-safe chunked processing.
Provides streaming CSV parse and generate patterns for NodeJS-Starter-V1, covering large file imports, exports, and transformations across Next.js frontend and FastAPI backend. Enforces row-by-row streaming, Zod and Pydantic row validation, and Australian locale formatting (DD/MM/YYYY dates, AUD currency).
When to Apply
Positive Triggers
Importing CSV files from user uploads
Exporting data to CSV for download (contractors, reports, agent runs)
Parsing large CSV datasets (1,000+ rows)
Validating CSV row structure against a schema
Transforming CSV data between formats
User mentions: "CSV", "import", "export", "spreadsheet", "download data", "upload file"
Negative Triggers
Validating form inputs without file upload (use data-validation instead)
Generating PDF or HTML reports (use report-generator patterns)
Processing JSON API responses (use instead)
api-contract
Working with binary file formats (xlsx, parquet) — different tooling required
Core Directives
The Three Laws of CSV Processing
Stream, never buffer: Process row-by-row. Never load entire files into memory.
Validate every row: Each row passes through a Zod (frontend) or Pydantic (backend) schema.
Locale-aware output: Dates as DD/MM/YYYY, currency as AUD, Australian English headers.
Recommended Libraries
Frontend (Next.js)
Library
Purpose
Install
papaparse
Streaming CSV parse (browser + Node)
pnpm add papaparse
@types/papaparse
TypeScript definitions
pnpm add -D @types/papaparse
Backend (FastAPI)
Library
Purpose
Install
python-multipart
File upload handling (already installed)
—
Built-in csv module
Streaming read/write
—
aiofiles
Async file I/O
uv add aiofiles
No additional backend library needed — Python's built-in csv module supports streaming via csv.reader and csv.DictWriter.
Wrap the parse function in a <input type="file" accept=".csv,text/csv"> component. Validate file extension (.csv), MIME type (text/csv), and size (default 10 MB cap) before parsing. Display row-level errors using ParseResult.errors.
Backend Patterns (FastAPI)
CSV Import Endpoint
import csv
import io
from typing importAnyfrom fastapi import APIRouter, File, HTTPException, UploadFile, status
from pydantic import BaseModel, Field, field_validator
router = APIRouter(prefix="/import", tags=["Import"])
classContractorImportRow(BaseModel):
"""Schema for a single CSV row."""
name: str = Field(min_length=1)
email: str
phone: str
abn: str
state: str @field_validator("phone") @classmethoddefvalidate_phone(cls, v: str) -> str:
import re
cleaned = re.sub(r"[^\d]", "", v)
ifnot re.match(r"^04\d{8}$", cleaned):
raise ValueError("Australian mobile required (04XX XXX XXX)")
return cleaned
classImportResult(BaseModel):
"""Result of CSV import."""
total_rows: int
valid_rows: int
error_rows: int
errors: list[dict[str, Any]] = Field(default_factory=list)
@router.post("/contractors", response_model=ImportResult)asyncdefimport_contractors(file: UploadFile = File(...)) -> ImportResult:
"""Import contractors from CSV file."""ifnot file.filename ornot file.filename.endswith(".csv"):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Only CSV files accepted",
)
content = await file.read()
reader = csv.DictReader(io.StringIO(content.decode("utf-8")))
valid: list[ContractorImportRow] = []
errors: list[dict[str, Any]] = []
for i, row inenumerate(reader, start=1):
try:
parsed = ContractorImportRow(**row)
valid.append(parsed)
except Exception as e:
errors.append({"row": i, "error": str(e)})
# Process valid rows (insert into database)# await bulk_insert_contractors(valid)return ImportResult(
total_rows=len(valid) + len(errors),
valid_rows=len(valid),
error_rows=len(errors),
errors=errors[:50], # Cap error list
)
CSV Export Endpoint
import csv
import io
from datetime import datetime
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
router = APIRouter(prefix="/export", tags=["Export"])
@router.get("/contractors")asyncdefexport_contractors() -> StreamingResponse:
"""Export contractors as CSV download."""# Fetch data
contractors = await get_all_contractors()
# Stream CSV output
output = io.StringIO()
writer = csv.DictWriter(
output,
fieldnames=["name", "email", "phone", "abn", "state", "created_at"],
)
writer.writeheader()
for c in contractors:
writer.writerow({
"name": c.name,
"email": c.email,
"phone": c.phone,
"abn": c.abn,
"state": c.state,
"created_at": _format_au_date(c.created_at),
})
output.seek(0)
timestamp = datetime.now().strftime("%d-%m-%Y")
return StreamingResponse(
iter([output.getvalue()]),
media_type="text/csv",
headers={
"Content-Disposition": f'attachment; filename="contractors-{timestamp}.csv"'
},
)
def_format_au_date(dt: datetime) -> str:
"""Format datetime as DD/MM/YYYY for CSV export."""return dt.strftime("%d/%m/%Y")
Streaming Large Files
For files over 10 MB, use async generators:
from collections.abc import AsyncGenerator
asyncdefstream_csv_rows(
file: UploadFile,
chunk_size: int = 8192,
) -> AsyncGenerator[dict[str, str], None]:
"""Stream CSV rows without loading entire file into memory."""
buffer = ""
reader = Nonewhile chunk := await file.read(chunk_size):
buffer += chunk.decode("utf-8")
lines = buffer.split("\n")
buffer = lines.pop() # Keep incomplete line in bufferif reader isNoneand lines:
# First chunk — extract headers
header_line = lines.pop(0)
headers = next(csv.reader([header_line]))
reader = headers
for line in lines:
if line.strip() and reader:
values = next(csv.reader([line]))
yielddict(zip(reader, values))
# Process remaining bufferif buffer.strip() and reader:
values = next(csv.reader([buffer]))
yielddict(zip(reader, values))
Australian Locale Formatting
Date Columns
Context
Format
Example
CSV export
DD/MM/YYYY
23/01/2026
CSV import (accept)
DD/MM/YYYY or ISO 8601
23/01/2026, 2026-01-23
Database storage
ISO 8601
2026-01-23T00:00:00Z
Date Parsing (Dual Format)
functionparseAustralianDate(value: string): Date {
// Try DD/MM/YYYY firstconst auMatch = value.match(/^(\d{2})\/(\d{2})\/(\d{4})$/);
if (auMatch) {
returnnewDate(`${auMatch[3]}-${auMatch[2]}-${auMatch[1]}`);
}
// Fall back to ISO 8601const iso = newDate(value);
if (!isNaN(iso.getTime())) return iso;
thrownewError(`Invalid date: ${value}`);
}
from datetime import datetime
defparse_au_date(value: str) -> datetime:
"""Parse DD/MM/YYYY or ISO 8601 date string."""for fmt in ("%d/%m/%Y", "%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"):
try:
return datetime.strptime(value, fmt)
except ValueError:
continueraise ValueError(f"Invalid date: {value}")
Currency Columns
// Export: format as AUDconst amount = 1234.5;
const formatted = `$${amount.toFixed(2)}`; // "$1234.50"// Import: strip $ and commasconst raw = '$1,234.50';
const parsed = parseFloat(raw.replace(/[$,]/g, '')); // 1234.5
