from sqlalchemy import select
from sqlalchemy.orm import Session
from .models import User
BATCH_SIZE = 1000
def stream_users(session: Session):
stmt = (
select(User.id, User.email, User.full_name, User.created_at)
.order_by(User.id)
.execution_options(yield_per=BATCH_SIZE)
)
# server-side batched fetch; rows arrive lazily, not all at once
for row in session.execute(stmt):
yield row
import csv
import io
from typing import Iterator
from sqlalchemy.orm import Session
from .repository import stream_users
HEADER = ["id", "email", "full_name", "created_at"]
def iter_csv(session: Session) -> Iterator[str]:
buffer = io.StringIO()
writer = csv.writer(buffer)
writer.writerow(HEADER)
yield buffer.getvalue()
buffer.seek(0)
buffer.truncate(0)
for row in stream_users(session):
writer.writerow([
row.id,
row.email,
row.full_name,
row.created_at.isoformat() if row.created_at else "",
])
yield buffer.getvalue()
buffer.seek(0)
buffer.truncate(0)
from datetime import date
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from .csv_stream import iter_csv
from .db import SessionLocal
router = APIRouter()
@router.get("/exports/users.csv")
def export_users() -> StreamingResponse:
session = SessionLocal()
def generate():
try:
yield from iter_csv(session)
finally:
session.close()
filename = f"users-{date.today().isoformat()}.csv"
headers = {
"Content-Disposition": f'attachment; filename="{filename}"',
}
return StreamingResponse(
generate(),
media_type="text/csv",
headers=headers,
)
This snippet shows how a FastAPI endpoint can export a very large table as CSV without buffering the whole file in memory. The core idea is that StreamingResponse accepts an iterator (or generator), and FastAPI drains it chunk by chunk onto the socket. As long as each chunk is produced lazily and the underlying data source is streamed too, memory stays flat regardless of whether the export is a thousand rows or ten million.
In csv_stream.py, the iter_csv generator is the heart of the technique. Instead of building a list of rows and calling writer.writerows, it wraps a small in-memory buffer with io.StringIO, writes a single logical row into it, yields the buffered text, then truncates the buffer with seek(0)/truncate(0). That truncation is what keeps the buffer bounded: the StringIO is reused as a formatting scratchpad rather than an accumulator. Using csv.writer matters here because it handles quoting, embedded commas, and newlines correctly — hand-rolling ','.join would corrupt data containing those characters.
The generator pulls its data from stream_users in repository.py, which uses SQLAlchemy's yield_per to fetch rows in server-side batches. This is critical: without it, the driver would materialize the entire result set before the first row is ever yielded, defeating the whole purpose. yield_per sets a fetch size so the database and generator stay in lockstep, giving natural backpressure — the query only advances as fast as the client consumes bytes.
In routes.py, the export_users endpoint returns a StreamingResponse built from iter_csv(session), with media_type set to text/csv and a Content-Disposition header so browsers download it as a file. Note there is no Content-Length; the response uses chunked transfer encoding because the total size is unknown up front.
One pitfall worth calling out: the database session must stay open for the entire duration of streaming, so the session is created inside the request and closed only after the generator is exhausted, not via a short-lived dependency. This pattern is the right reach whenever an export is large enough that loading it fully would risk memory pressure or slow time-to-first-byte.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
package deps
import (
"net"
"net/http"
"time"HTTP client tuned for production: timeouts, transport, and connection reuse
package api
import (
"io"
"net/http"
"os"Safe multipart uploads using temp files (bounded memory)
Share this code
Here's the card — post it anywhere.
