CSV из FastAPI: StreamingResponse без сборки всего файла в строку

Отчёт сначала складывается в список, затем в огромную строку CSV и только после этого отправляется клиенту. Для большого экспорта это лишние копии данных. StreamingResponse может отдавать результат генератора по частям, а csv.writer правильно обработает запятые, кавычки и переводы строк.

Генератор строит одну CSV-строку за раз

Нельзя собирать CSV через join по запятой: имя с запятой превратится в две колонки, а перевод строки испортит структуру записи. csv.writer знает правила экранирования. Небольшой StringIO переиспользуется для одной строки, после yield очищается. Память такого преобразования зависит от текущей строки, а не от полной текстовой копии экспорта.

Чтобы экономия сохранилась, источник тоже должен поступать постепенно. list(db_result) перед генератором всё равно загрузит записи целиком. Для базы выберите подходящее пакетное чтение, стабильный порядок и границы Session. Не рассчитывайте на закрытую зависимость, если генератор ещё читает отношения ORM.

Проверим CSV обычным парсером

Нужны FastAPI и HTTPX. Источник из двух строк специально содержит запятую и кавычки. Тест читает HTTP-ответ и разбирает его csv.reader, сравнивая исходные значения. Это подтверждает формат CSV, но не измеряет поминутную сетевую доставку: TestClient собирает ответ для проверки.

Сохраните весь блок в файл и запустите Python 3.11+ в отдельном окружении с пакетами fastapi, httpx. Для моделей используется Pydantic v2; отдельный сервер для TestClient не нужен.

import csv
import io
from itertools import chain
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient

app = FastAPI()
records = [(1, "Room, East"), (2, 'Room "Blue"')]

def csv_chunks():
    buffer = io.StringIO(newline="")
    writer = csv.writer(buffer)
    for row in chain([("id", "name")], records):
        writer.writerow(row)
        yield buffer.getvalue()
        buffer.seek(0)
        buffer.truncate(0)

@app.get("/export.csv")
def export():
    return StreamingResponse(csv_chunks(), media_type="text/csv",
        headers={"Content-Disposition": 'attachment; filename="rooms.csv"'})

with TestClient(app) as client:
    response = client.get("/export.csv")
    assert response.status_code == 200
    assert response.headers["content-type"].startswith("text/csv")
    assert "rooms.csv" in response.headers["content-disposition"]
    parsed = list(csv.reader(io.StringIO(response.text)))
    assert parsed == [["id", "name"], ["1", "Room, East"], ["2", 'Room "Blue"']]
    assert len(list(csv_chunks())) == 3
print("CSV rows decoded correctly:", parsed)

Не потеряйте поток на стороне источника

Учебный источник здесь мал и уже находится в памяти. При замене на базу передавайте итератор строк без преобразования всего результата в список. Заголовок можно отдать отдельно, а затем выполнить for row in result. Важно изменить и получение данных, и формирование CSV: одно StreamingResponse не превращает готовый большой список в потоковую выборку.

Исключение генератора после отправки заголовков уже не всегда можно превратить в аккуратный JSON 500. Поэтому разрешение доступа и базовые параметры отчёта проверяйте до возврата StreamingResponse. Для экспорта, который должен завершаться атомарно, иногда удобнее заранее создать файл фоновой задачей и отдать его после успешной подготовки.

Учтите программу, которая откроет файл

Для текстового обмена UTF-8 обычно удобен. Если целевой табличный редактор требует BOM или иной разделитель, согласуйте это как часть формата и добавьте соответствующую проверку байтов. CSV-формулы тоже требуют отдельного решения: значения, начинающиеся с =, +, - или @, некоторые редакторы исполняют как формулы. Экранирование CSV само по себе не нейтрализует такой смысл.

Для сетевой проверки используйте тестовый сервер и клиент, который читает поток кусками, если нужно подтвердить раннюю отдачу или поведение прокси. Локальная проверка выше решает другую конкретную задачу: корректные поля и HTTP-заголовки выгрузки. Оба свойства нужны, но измеряются разными способами.

Следующая практика

В открытом уроке TestClient вы научитесь проверять реальный HTTP-ответ. Используйте эту основу для CSV: сравните тип содержимого, заголовок скачивания и значения после обратного разбора файла. Откройте бесплатный урок проверки ответа API.

Бесплатная программа FastAPI поможет построить источник данных для экспорта: API с SQLAlchemy и тестами. Когда каталог хранится в базе, выбирайте объём и порядок выгрузки по реальной задаче отчёта.