python 48 lines · 8 steps

Streaming a CSV export in FastAPI

An async generator pairs SQLAlchemy's streaming queries with FastAPI's StreamingResponse to export orders without loading them all into memory.

Explained by highlit
FastAPI Advanced streaming async-generators csv memory-efficiency database-streaming
1import csv
2import io
3from datetime import date
4 
5from fastapi import APIRouter, Depends
6from fastapi.responses import StreamingResponse
7from sqlalchemy import select
8from sqlalchemy.ext.asyncio import AsyncSession
9 
10from app.db import get_session
11from app.models import Order
12 
13router = APIRouter(prefix="/exports", tags=["exports"])
14 
15 
16async def _stream_orders(session: AsyncSession):
17 buffer = io.StringIO()
18 writer = csv.writer(buffer)
19 
20 writer.writerow(["id", "customer_email", "total_cents", "status", "created_at"])
21 yield buffer.getvalue()
22 buffer.seek(0)
23 buffer.truncate(0)
24 
25 stmt = select(Order).order_by(Order.id).execution_options(yield_per=500)
26 result = await session.stream(stmt)
27 
28 async for order in result.scalars():
29 writer.writerow([
30 order.id,
31 order.customer_email,
32 order.total_cents,
33 order.status,
34 order.created_at.isoformat(),
35 ])
36 yield buffer.getvalue()
37 buffer.seek(0)
38 buffer.truncate(0)
39 
40 
41@router.get("/orders.csv")
42async def export_orders(session: AsyncSession = Depends(get_session)):
43 filename = f"orders-{date.today():%Y%m%d}.csv"
44 return StreamingResponse(
45 _stream_orders(session),
46 media_type="text/csv",
47 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
48 )
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1An async generator lets you produce a response incrementally, keeping peak memory flat regardless of row count.
  2. 2Reusing one StringIO buffer with seek/truncate avoids accumulating the entire CSV in memory.
  3. 3SQLAlchemy's stream with yield_per fetches rows in batches instead of materializing the full result set.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
python 
import re
from functools import total_ordering
from typing import Optional
 

Parsing and comparing semantic versions

regex operator-overloading sorting
Intermediate 7 steps
python 
from typing import Any, Sequence, Mapping
 
 
def render_markdown_table(

Rendering an aligned Markdown table in Python

string formatting data transformation closures
Intermediate 8 steps

Share this explainer

Here's the card — post it anywhere.

Streaming a CSV export in FastAPI — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code