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

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

Related explainers

python

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status
from pydantic import BaseModel, EmailStr
from sqlalchemy.orm import Session

Building a signup endpoint in FastAPI

dependency-injection request-validation background-tasks

Intermediate 8 steps

python

from django import forms
from django.utils import timezone
 
from .models import Reservation

Multi-field validation in a Django ModelForm

form validation cross-field validation modelform

Intermediate 7 steps

python

from django import template
from django.urls import reverse, NoReverseMatch
from django.utils.html import format_html

Active nav-link template tags in Django

template tags url routing active state

Intermediate 7 steps

python

from functools import wraps
import asyncio
 
from fastapi import APIRouter, FastAPI, Request

Per-route request timeouts in FastAPI

decorators async timeouts

Intermediate 6 steps

python

import os
from pathlib import Path
 
BASE_DIR = Path(__file__).resolve().parent.parent.parent

How a Django settings module is wired

configuration environment-variables middleware

Intermediate 8 steps

func UploadChunk(c *gin.Context) {
	uploadID := c.Param("uploadID")
	if !validUploadID.MatchString(uploadID) {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid upload id"})