Streaming query results as CSV from FastAPI

Return a StreamingResponse whose generator opens its own AsyncSession and reads with session.stream() plus yield_per — the request's session is already closed by the time the body is produced, and buffering the whole export in memory both delays the first byte and risks the process. This guide belongs to streaming large result sets with yield_per.

Quick Answer

A streaming response body is produced after the endpoint returns, so the session a dependency provided is gone by then.

Before — the whole export in memory, on the request's session:

import csv
import io

from fastapi import Depends, FastAPI
from fastapi.responses import Response
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from shop.db import get_session
from shop.models import Order

app = FastAPI()


@app.get("/orders.csv")
async def export_orders(session: AsyncSession = Depends(get_session)) -> Response:
    rows = (await session.execute(select(Order.id, Order.placed_on, Order.total_cents))).all()
    buffer = io.StringIO()
    writer = csv.writer(buffer)
    writer.writerow(["id", "placed_on", "total_cents"])
    writer.writerows(rows)                     # every row in memory, twice
    return Response(buffer.getvalue(), media_type="text/csv")

After — streamed in batches, on a session the generator owns:

import csv
import io
from collections.abc import AsyncIterator

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from sqlalchemy import select

from shop.db import Session
from shop.models import Order

app = FastAPI()
COLUMNS = ["id", "placed_on", "total_cents"]


async def order_rows() -> AsyncIterator[str]:
    buffer = io.StringIO(newline="")
    writer = csv.writer(buffer)

    def flush() -> str:
        value = buffer.getvalue()
        buffer.seek(0)
        buffer.truncate(0)
        return value

    writer.writerow(COLUMNS)
    yield "\ufeff" + flush()                   # BOM first, for Excel on Windows

    async with Session() as session:            # the generator owns its session
        result = await session.stream(
            select(Order.id, Order.placed_on, Order.total_cents)
            .order_by(Order.id)
            .execution_options(yield_per=1_000)
        )
        async for batch in result.partitions(1_000):
            writer.writerows(batch)
            yield flush()


@app.get("/orders.csv")
async def export_orders() -> StreamingResponse:
    return StreamingResponse(
        order_rows(),
        media_type="text/csv; charset=utf-8",
        headers={"Content-Disposition": 'attachment; filename="orders.csv"'},
    )

Peak memory is one batch, the first bytes reach the client immediately, and the session lives exactly as long as the iteration.

Execution Context & Async Workflow Integration

A StreamingResponse stores the async generator and returns. FastAPI then finishes the dependency stack — which closes any session a Depends provided — and Starlette begins iterating the generator to write the response body. Every database call in the generator therefore happens after the request scope has ended.

That is the same boundary background tasks cross, with the same consequences: a session borrowed from the request raises, and objects loaded through it are detached. The generator has to open its own session, and because it is an async with block inside the generator, the session closes when iteration finishes — including when it finishes early.

session.stream() is what keeps memory flat. It sets stream_results, so asyncpg uses a server-side cursor and rows arrive in batches rather than all at once, and yield_per sets the batch size. result.partitions(n) then yields lists of rows, which suits csv.writer.writerows() exactly: one formatting call per batch rather than per row.

The buffer trick in the example matters for memory. csv.writer needs a file-like object, and a StringIO that is never truncated grows to the size of the export — which defeats the purpose. Reading its value, seeking to zero and truncating after each batch keeps it at one batch's worth.

Three operational consequences are worth planning for.

A connection is held for the whole download. A server-side cursor lives inside a transaction, so the connection cannot be returned until iteration ends. Ten concurrent exports from a pool of ten leave nothing for ordinary requests, which is an argument for a separate, small engine for exports — the isolation described in running background tasks with a fresh AsyncSession in FastAPI.

A client disconnect cancels the generator. Starlette raises cancellation into it, the async with block closes the session, and the connection returns to the pool. Without a context manager — a session opened and closed by hand — that path leaks, which is the failure described in fixing garbage collector non-checked-in connection warnings.

A long-running transaction holds back vacuum. An export that streams for twenty minutes keeps a snapshot open for twenty minutes, during which dead rows from other transactions cannot be cleaned up. For very large exports that is a reason to prefer batched, separately committed reads — the subject of processing large tables in batches with partitions.

Resolving Warnings, Errors & Common Mistakes

Two of these are worth expanding.

The newline detail catches everyone once. csv.writer emits \r\n by default, and a text buffer that performs newline translation turns that into \r\r\n. Passing newline="" to io.StringIO disables translation, which is the same requirement the csv documentation states for files.

The connection-holding problem is the one that turns a working feature into an incident under load. An export engine keeps it contained:

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

export_engine = create_async_engine(
    DATABASE_URL,
    pool_size=2,
    max_overflow=1,
    pool_timeout=30,
    connect_args={"server_settings": {
        "application_name": "shop-export",
        "statement_timeout": "0",        # exports may legitimately run long
    }},
)
ExportSession = async_sessionmaker(export_engine, expire_on_commit=False)

Three connections maximum means at most three simultaneous exports, a fourth request waits rather than starving the API, and pg_stat_activity shows exports separately from request traffic. Disabling statement_timeout for this engine only is the other half: an export is one long statement by design, and the API's five-second limit would kill it.

Advanced: Formats, Compression and Very Large Exports

CSV is the format users ask for; two variations are worth supporting when the data is large.

Gzip on the fly reduces transfer size by an order of magnitude for tabular text, and costs a little CPU per batch:

import gzip
import io
from collections.abc import AsyncIterator


async def gzipped(rows: AsyncIterator[str]) -> AsyncIterator[bytes]:
    buffer = io.BytesIO()
    compressor = gzip.GzipFile(fileobj=buffer, mode="wb")
    try:
        async for chunk in rows:
            compressor.write(chunk.encode("utf-8"))
            compressor.flush()
            data = buffer.getvalue()
            if data:
                buffer.seek(0)
                buffer.truncate(0)
                yield data
    finally:
        compressor.close()
    yield buffer.getvalue()

Serve it with Content-Encoding: gzip and a .csv.gz filename. Many gateways compress responses already, in which case doing it twice is wasted work — check before adding it.

JSON Lines is often a better interface for machine consumers: one JSON object per line, no quoting rules, and streaming-friendly:

import json


async def order_lines() -> AsyncIterator[str]:
    async with Session() as session:
        result = await session.stream(
            select(Order.id, Order.placed_on, Order.total_cents)
            .order_by(Order.id).execution_options(yield_per=1_000)
        )
        async for batch in result.partitions(1_000):
            yield "".join(
                json.dumps({"id": i, "placed_on": d.isoformat(), "total_cents": t}) + "\n"
                for i, d, t in batch
            )

Past a certain size, streaming from a request stops being the right shape at all. The signals are concrete: exports that approach the gateway's total-duration limit, users retrying because a download failed at ninety percent, or connection pressure from concurrent exports. The alternative is a job that writes to object storage and a link the user fetches:

from shop.jobs import queue_export


@app.post("/orders/export")
async def request_export(customer_id: int) -> dict:
    job_id = await queue_export(customer_id)
    return {"job_id": job_id, "status_url": f"/exports/{job_id}"}

That removes every request-path constraint — timeouts, connection holding, retries, resumable downloads — at the cost of a worker, a storage bucket and a status endpoint. The worker itself should use the batched approach rather than one long transaction, and PostgreSQL's COPY ... TO is worth knowing there: it writes CSV server-side far faster than any row-by-row loop, and the client-side equivalent is described in loading rows with Postgres COPY through asyncpg.

Testing a Streaming Endpoint

Streaming endpoints have two failure modes that ordinary endpoint tests miss: the generator running after the session closed, and memory growing with the export. Both are testable.

Assert the whole body, and that it took more than one chunk. httpx with ASGITransport iterates the response like a client would:

import csv
import io

import httpx
import pytest

from shop.main import app


@pytest.mark.asyncio
async def test_export_streams_all_rows(seeded_orders):
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        async with client.stream("GET", "/orders.csv") as response:
            assert response.status_code == 200
            assert response.headers["content-type"].startswith("text/csv")
            assert "attachment" in response.headers["content-disposition"]
            chunks = [chunk async for chunk in response.aiter_text()]

    body = "".join(chunks).lstrip("\ufeff")
    rows = list(csv.reader(io.StringIO(body)))
    assert rows[0] == ["id", "placed_on", "total_cents"]
    assert len(rows) == len(seeded_orders) + 1
    assert len(chunks) > 1, "the response was buffered rather than streamed"

The last assertion is the one that catches a regression from streaming back to buffering — for instance, a well-meaning change that collects the generator into a list before returning it.

Test the disconnect path. Breaking out of the iteration mid-download should leave nothing checked out:

@pytest.mark.asyncio
async def test_client_disconnect_releases_the_connection(seeded_orders, engine):
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        async with client.stream("GET", "/orders.csv") as response:
            async for _ in response.aiter_bytes():
                break                              # disconnect after the first chunk

    assert engine.sync_engine.pool.checkedout() == 0

Without an async with Session() inside the generator, that assertion fails — which is the proof that the session ownership is right.

Seed enough rows to cross a batch boundary. A test with five rows and yield_per=1_000 exercises a single partition and would pass even if batching were broken. Seeding two or three thousand rows, or lowering yield_per in a test-specific override, makes the batching path real. The fixtures for either approach are the ones in using pytest-asyncio fixtures with AsyncSession.

Frequently Asked Questions

Related

• Streaming Large Result Sets with yield_per — The parent guide: server-side cursors and memory.
• Processing large tables in batches with partitions — Batched reads that commit as they go.
• Using yield_per to stream millions of rows in async — How streaming works underneath.
• Running background tasks with a fresh AsyncSession in FastAPI — The same session-ownership boundary.