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 
import argparse
import sys
from pathlib import Path
 

Building a subcommand CLI with argparse

cli argparse subcommands
Intermediate 6 steps
python 
from collections.abc import Mapping
from typing import Any, Iterator
 
 

Flattening nested config into dotted keys

recursion generators tree-traversal
Intermediate 7 steps
python 
import csv
import io
from datetime import datetime
 

Streaming a CSV export in Flask

streaming generators csv
Intermediate 9 steps
python 
import time
from collections import defaultdict
from threading import Lock
 

Sliding-window login rate limiting in Flask

rate-limiting sliding-window thread-safety
Intermediate 7 steps
python 
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.mail import EmailMultiAlternatives
from django.db.models.signals import post_save

Sending a welcome email with Django signals

signals email user-activation
Intermediate 8 steps
go 
func (h *TransactionHandler) ExportCSV(c *gin.Context) {
	ctx := c.Request.Context()
	filters := parseTransactionFilters(c)
 

Streaming a CSV export in Gin

streaming csv-export database-cursor
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