Streaming and File Responses

Capsule overview

So far, every one of your responses has been JSON: FastAPI serializes a dict or Pydantic model, turns it into JSON, and sends the whole thing to the client. That works perfectly for small responses — one task, a list of 50 items. But what happens when you need to export 100,000 records as CSV? Or send a 500MB log file? Loading everything into memory before sending it isn't viable.

In this capsule you learn two essential tools: StreamingResponse and FileResponse. StreamingResponse uses generators (yield) to send data chunk by chunk — the server never loads everything into memory. FileResponse serves files from disk with the right headers so the browser downloads them. Together, these tools let you handle large data and files the way professional APIs do.

The problem with large responses

Look at what happens when you try to export a lot of data as conventional JSON:

from fastapi import FastAPI

app = FastAPI()


def generate_large_dataset(n: int) -> list[dict]:
    """Generates N records in memory."""
    return [
        {"id": i, "name": f"Item {i}", "value": i * 1.5}
        for i in range(1, n + 1)
    ]


@app.get("/export-json")
def export_json():
    data = generate_large_dataset(100_000)
    return data

The problem: generate_large_dataset(100_000) creates 100,000 dictionaries in memory before returning anything. With bigger datasets, your server runs out of memory and crashes. And the client has to wait for the whole list to be generated before receiving the first byte.

StreamingResponse: sending data chunk by chunk

StreamingResponse solves the problem with a generator — a function that produces data incrementally with yield:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import io

app = FastAPI()

tasks_data = [
    {"id": 1, "title": "Set up CI/CD", "status": "in_progress", "priority": 4},
    {"id": 2, "title": "Write tests", "status": "pending", "priority": 3},
    {"id": 3, "title": "Document the API", "status": "completed", "priority": 2},
    {"id": 4, "title": "Code review", "status": "pending", "priority": 5},
    {"id": 5, "title": "Deploy staging", "status": "in_progress", "priority": 4},
]


def csv_generator(data: list[dict]):
    """Generator that produces CSV line by line."""
    if not data:
        return

    headers = data[0].keys()
    yield ",".join(headers) + "\n"

    for row in data:
        values = [str(row[h]) for h in headers]
        yield ",".join(values) + "\n"


@app.get("/tasks/export")
def export_tasks_csv():
    return StreamingResponse(
        content=csv_generator(tasks_data),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=tasks.csv"},
    )

curl http://127.0.0.1:8000/tasks/export

id,title,status,priority
1,Set up CI/CD,in_progress,4
2,Write tests,pending,3
3,Document the API,completed,2
4,Code review,pending,5
5,Deploy staging,in_progress,4

How does it work?

1. csv_generator is a generator function — it uses yield instead of return
2. Every yield sends one CSV line to the client immediately
3. The function pauses after each yield and picks back up when the next chunk is needed
4. Only one line is in memory at a time — it doesn't matter whether there are 10 or 10 million records

The Content-Disposition header

headers={"Content-Disposition": "attachment; filename=tasks.csv"}

This header tells the browser: "This isn't content to display — it's a file to download, and its suggested name is tasks.csv." Without it, the browser would show the CSV as plain text in the tab.

Streaming with large datasets

The real power of streaming shows up with data that doesn't fit in memory:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from datetime import datetime, timedelta
import random

app = FastAPI()


def generate_large_csv(num_records: int):
    """Generator for a CSV with many records."""
    yield "id,title,status,priority,created_at\n"

    statuses = ["pending", "in_progress", "completed"]
    base_date = datetime(2026, 1, 1)

    for i in range(1, num_records + 1):
        status = random.choice(statuses)
        priority = random.randint(1, 5)
        created = base_date + timedelta(hours=i)
        yield f"{i},Task {i},{status},{priority},{created.isoformat()}\n"


@app.get("/tasks/export/large")
def export_large_csv(count: int = 10000):
    """Exports up to 100K tasks as streaming CSV."""
    count = min(count, 100_000)

    return StreamingResponse(
        content=generate_large_csv(count),
        media_type="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename=tasks_{count}.csv",
        },
    )

curl "http://127.0.0.1:8000/tasks/export/large?count=50000" -o tasks.csv

This generates and sends 50,000 CSV lines. The server never holds all 50,000 lines in memory at once — just one at a time. The client starts receiving data immediately, without waiting for everything to be generated.

Streaming with Python's csv module

For more robust CSV (fields with commas, quotes, special characters), use the standard library's csv module:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import csv
import io

app = FastAPI()

tasks_data = [
    {"id": 1, "title": "Set up CI/CD", "status": "in_progress", "assignee": "Ana García"},
    {"id": 2, "title": 'Review "edge cases"', "status": "pending", "assignee": "Carlos, Jr."},
    {"id": 3, "title": "Deploy v2.0", "status": "completed", "assignee": "María López"},
]


def csv_generator_robust(data: list[dict]):
    """Generator that produces CSV with the csv module (handles quotes and commas)."""
    if not data:
        return

    output = io.StringIO()
    writer = csv.DictWriter(output, fieldnames=data[0].keys())

    writer.writeheader()
    yield output.getvalue()
    output.seek(0)
    output.truncate(0)

    for row in data:
        writer.writerow(row)
        yield output.getvalue()
        output.seek(0)
        output.truncate(0)


@app.get("/tasks/export/csv")
def export_tasks_robust():
    return StreamingResponse(
        content=csv_generator_robust(tasks_data),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=tasks.csv"},
    )

curl http://127.0.0.1:8000/tasks/export/csv

id,title,status,assignee
1,Set up CI/CD,in_progress,Ana García
2,"Review ""edge cases""",pending,"Carlos, Jr."
3,Deploy v2.0,completed,María López

The csv module automatically escapes quotes ("edge cases" → ""edge cases"") and wraps fields containing commas (Carlos, Jr. → "Carlos, Jr.").

Streaming logs in real time

StreamingResponse is also useful for sending data incrementally, like logs:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time
from datetime import datetime

app = FastAPI()


def log_stream_generator(lines: int = 20):
    """Simulates a real-time log stream."""
    log_levels = ["INFO", "DEBUG", "WARNING", "ERROR"]
    messages = [
        "Request processed successfully",
        "Database query executed in 23ms",
        "Cache miss for key user:42",
        "Connection pool at 80% capacity",
        "Rate limit approaching for IP 192.168.1.100",
    ]

    for i in range(lines):
        timestamp = datetime.now().isoformat()
        level = log_levels[i % len(log_levels)]
        message = messages[i % len(messages)]
        yield f"[{timestamp}] {level}: {message}\n"
        time.sleep(0.1)


@app.get("/logs/stream")
def stream_logs(lines: int = 20):
    """A simulated log stream."""
    return StreamingResponse(
        content=log_stream_generator(min(lines, 100)),
        media_type="text/plain",
    )

curl http://127.0.0.1:8000/logs/stream?lines=5

The logs show up line by line with a 100ms delay between each one. The time.sleep(0.1) simulates logs being generated in real time.

FileResponse: serving files from disk

FileResponse serves files that already exist on the server's filesystem:

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pathlib import Path

app = FastAPI()

REPORTS_DIR = Path("reports")
REPORTS_DIR.mkdir(exist_ok=True)

report_content = """# Task Report - March 2026

## Summary
- Total tasks: 42
- Completed: 28
- In progress: 10
- Pending: 4

## Top Priority Tasks
1. Deploy production v3.0
2. Security audit
3. Performance optimization
"""
report_path = REPORTS_DIR / "task-report-2026-03.md"
report_path.write_text(report_content)


@app.get("/reports/{filename}")
def download_report(filename: str):
    file_path = REPORTS_DIR / filename

    if not file_path.exists():
        raise HTTPException(status_code=404, detail=f"Report '{filename}' not found")

    if not file_path.is_relative_to(REPORTS_DIR):
        raise HTTPException(status_code=403, detail="Access denied")

    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="application/octet-stream",
    )

curl http://127.0.0.1:8000/reports/task-report-2026-03.md -O

FileResponse parameters

Security: validating file paths

The file_path.is_relative_to(REPORTS_DIR) line prevents path traversal attacks. Without it, a client could ask for /reports/../../etc/passwd and reach system files.

# ❌ No validation — vulnerable to path traversal
@app.get("/files/{filename}")
def get_file(filename: str):
    return FileResponse(f"files/{filename}")

# ✅ With validation — only files inside the allowed directory
@app.get("/files/{filename}")
def get_file(filename: str):
    file_path = Path("files") / filename
    if not file_path.exists() or not file_path.is_relative_to(Path("files")):
        raise HTTPException(status_code=404, detail="File not found")
    return FileResponse(file_path)

FileResponse with on-demand generation

You can generate a file, serve it, and clean it up:

from fastapi import FastAPI
from fastapi.responses import FileResponse
from pathlib import Path
import csv
import tempfile

app = FastAPI()

tasks_data = [
    {"id": 1, "title": "Set up CI/CD", "status": "in_progress", "priority": 4},
    {"id": 2, "title": "Write tests", "status": "pending", "priority": 3},
    {"id": 3, "title": "Document the API", "status": "completed", "priority": 2},
]


@app.get("/tasks/report")
def generate_task_report():
    """Generates a temporary CSV file and serves it as a download."""
    temp_file = tempfile.NamedTemporaryFile(
        mode="w", suffix=".csv", delete=False, encoding="utf-8",
    )

    writer = csv.DictWriter(temp_file, fieldnames=["id", "title", "status", "priority"])
    writer.writeheader()
    writer.writerows(tasks_data)
    temp_file.close()

    return FileResponse(
        path=temp_file.name,
        filename="task-report.csv",
        media_type="text/csv",
    )

tempfile.NamedTemporaryFile(delete=False) creates a temporary file that sticks around until you delete it explicitly. FastAPI reads it and sends it to the client. In production you'd want a background task to clean up temporary files (you'll see that in Module 4).

StreamingResponse vs FileResponse

Rule of thumb:

• If you generate the data → StreamingResponse with a generator
• If the file already exists → FileResponse
• If you need to generate and then serve → generate + FileResponse (or StreamingResponse directly)

Streaming with async generators

If your data comes from an async source (a database, an external API), use an async generator:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()


async def async_csv_generator(num_records: int):
    """Async generator that simulates data coming from a database."""
    yield "id,title,status\n"

    for i in range(1, num_records + 1):
        await asyncio.sleep(0.01)
        yield f"{i},Task {i},pending\n"


@app.get("/tasks/export/async")
async def export_tasks_async(count: int = 100):
    return StreamingResponse(
        content=async_csv_generator(min(count, 10000)),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=tasks_async.csv"},
    )

The await asyncio.sleep(0.01) simulates a database query. In a real case, you'd do await db.fetch() for each batch of records. The async generator lets the event loop serve other requests while it waits for the data.

Common media types

Connection with the project

In the project capsule (05), you'll implement a GET /tasks/export that exports all your tasks as CSV using StreamingResponse, and a GET /tasks/{id}/report that generates and serves a report as a FileResponse. These endpoints tie into the Module 3 schemas — the CSV uses TaskPublic's fields, not TaskAdmin's.

Troubleshooting

Problem 1: The CSV file displays in the browser instead of downloading

Cause: The Content-Disposition header is missing.

Fix:

# ❌ No Content-Disposition — the browser shows the CSV as text
return StreamingResponse(content=csv_gen(), media_type="text/csv")

# ✅ With Content-Disposition — the browser opens a download dialog
return StreamingResponse(
    content=csv_gen(),
    media_type="text/csv",
    headers={"Content-Disposition": "attachment; filename=data.csv"},
)

Problem 2: StreamingResponse sends everything at once — no actual streaming

Cause: The generator builds everything before yielding. If your generator has a return [all_data] before the yield loop, there's no real streaming.

Fix: Make sure the generator produces one chunk at a time:

# ❌ Not streaming — it builds everything in memory first
def bad_generator(data):
    all_lines = []
    for row in data:
        all_lines.append(f"{row['id']},{row['title']}\n")
    yield "".join(all_lines)

# ✅ Real streaming — one line at a time
def good_generator(data):
    for row in data:
        yield f"{row['id']},{row['title']}\n"

Problem 3: FileResponse returns a 500 — "File not found"

Cause: The path to the file is wrong. FileResponse expects an absolute path, or one relative to the server's working directory.

Fix:

from pathlib import Path

# ❌ A relative path that may not exist
return FileResponse("reports/data.csv")

# ✅ Check that it exists first
file_path = Path("reports/data.csv")
if not file_path.exists():
    raise HTTPException(status_code=404, detail="File not found")
return FileResponse(file_path)

Problem 4: A CSV with special characters (accents, ñ) looks garbled

Cause: Wrong encoding. The file is generated in UTF-8 but the client reads it as Latin-1.

Fix: Add a BOM (Byte Order Mark) at the start of the CSV so Excel reads it as UTF-8:

def csv_generator_utf8(data):
    yield "\ufeff"  # BOM for UTF-8
    yield ",".join(data[0].keys()) + "\n"
    for row in data:
        yield ",".join(str(v) for v in row.values()) + "\n"

Problem 5: The generator is exhausted after the first request

Cause: Generators in Python are exhausted after you iterate them once. If you store the generator in a variable and reuse it, the second request gets empty data.

Fix: Create a new generator on every request:

# ❌ A shared generator — exhausted after the first request
gen = csv_generator(data)

@app.get("/export")
def export():
    return StreamingResponse(gen)  # empty after the first request

# ✅ A new generator per request
@app.get("/export")
def export():
    return StreamingResponse(csv_generator(data))  # fresh every time

Exercises

Exercise 1: A basic CSV export (Easy)

Create a list of 6 products (id, name, price, category) and a GET /products/export endpoint that returns a StreamingResponse with CSV, including the header row and the Content-Disposition header for download.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

products = [
    {"id": 1, "name": "Laptop Pro", "price": 1299.99, "category": "Electronics"},
    {"id": 2, "name": "Mouse Wireless", "price": 29.99, "category": "Electronics"},
    {"id": 3, "name": "Python Book", "price": 35.00, "category": "Books"},
    {"id": 4, "name": "Standing Desk", "price": 450.00, "category": "Furniture"},
    {"id": 5, "name": "Monitor 4K", "price": 499.99, "category": "Electronics"},
    {"id": 6, "name": "Mechanical Keyboard", "price": 89.99, "category": "Electronics"},
]


def products_csv_generator():
    yield "id,name,price,category\n"
    for p in products:
        yield f"{p['id']},{p['name']},{p['price']},{p['category']}\n"


@app.get("/products/export")
def export_products():
    return StreamingResponse(
        content=products_csv_generator(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=products.csv"},
    )

curl http://127.0.0.1:8000/products/export
# → id,name,price,category
#   1,Laptop Pro,1299.99,Electronics
#   ...

Exercise 2: CSV with a query-parameter filter (Easy)

Extend the previous exercise: add a category query parameter to the export endpoint. If it's provided, export only products in that category.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from typing import Optional

app = FastAPI()

products = [
    {"id": 1, "name": "Laptop Pro", "price": 1299.99, "category": "Electronics"},
    {"id": 2, "name": "Mouse Wireless", "price": 29.99, "category": "Electronics"},
    {"id": 3, "name": "Python Book", "price": 35.00, "category": "Books"},
    {"id": 4, "name": "Standing Desk", "price": 450.00, "category": "Furniture"},
    {"id": 5, "name": "Monitor 4K", "price": 499.99, "category": "Electronics"},
    {"id": 6, "name": "Mechanical Keyboard", "price": 89.99, "category": "Electronics"},
]


def products_csv_generator(data: list[dict]):
    yield "id,name,price,category\n"
    for p in data:
        yield f"{p['id']},{p['name']},{p['price']},{p['category']}\n"


@app.get("/products/export")
def export_products(category: Optional[str] = None):
    data = products
    if category:
        data = [p for p in products if p["category"].lower() == category.lower()]

    filename = f"products_{category or 'all'}.csv"
    return StreamingResponse(
        content=products_csv_generator(data),
        media_type="text/csv",
        headers={"Content-Disposition": f"attachment; filename={filename}"},
    )

curl "http://127.0.0.1:8000/products/export?category=Electronics"
# → Only Electronics products

Exercise 3: FileResponse with validation (Medium)

Create a docs/ directory with 3 text files (.txt) created at app startup. Implement GET /docs/{filename} that serves files with FileResponse. Validate that the file exists and that the extension is .txt (reject other types with a 400).

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pathlib import Path

app = FastAPI()

DOCS_DIR = Path("docs")
DOCS_DIR.mkdir(exist_ok=True)

(DOCS_DIR / "guide.txt").write_text("FastAPI Advanced Features Guide\nModule 3: Response Models")
(DOCS_DIR / "changelog.txt").write_text("v3.0 - Added streaming\nv2.0 - Added routers")
(DOCS_DIR / "notes.txt").write_text("Remember to add error handling\nTest all endpoints")


@app.get("/docs/{filename}")
def download_doc(filename: str):
    if not filename.endswith(".txt"):
        raise HTTPException(
            status_code=400,
            detail="Only .txt files are allowed",
        )

    file_path = DOCS_DIR / filename

    if not file_path.exists():
        raise HTTPException(status_code=404, detail=f"File '{filename}' not found")

    if not file_path.is_relative_to(DOCS_DIR):
        raise HTTPException(status_code=403, detail="Access denied")

    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="text/plain",
    )

curl http://127.0.0.1:8000/docs/guide.txt
# → The file's contents

curl http://127.0.0.1:8000/docs/secret.pdf
# → 400: "Only .txt files are allowed"

curl http://127.0.0.1:8000/docs/missing.txt
# → 404: "File 'missing.txt' not found"

Exercise 4: Streaming with a configurable format (Medium)

Create a GET /tasks/export endpoint that accepts a format query parameter with the values "csv" or "jsonl" (JSON Lines — one JSON object per line). Use StreamingResponse with a different generator depending on the format.

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import StreamingResponse
from typing import Literal
import json

app = FastAPI()

tasks = [
    {"id": 1, "title": "Setup CI/CD", "status": "in_progress", "priority": 4},
    {"id": 2, "title": "Write tests", "status": "pending", "priority": 3},
    {"id": 3, "title": "Document API", "status": "completed", "priority": 2},
    {"id": 4, "title": "Code review", "status": "pending", "priority": 5},
]


def csv_generator(data: list[dict]):
    yield "id,title,status,priority\n"
    for task in data:
        yield f"{task['id']},{task['title']},{task['status']},{task['priority']}\n"


def jsonl_generator(data: list[dict]):
    for task in data:
        yield json.dumps(task, ensure_ascii=False) + "\n"


FORMAT_CONFIG = {
    "csv": {"generator": csv_generator, "media_type": "text/csv", "extension": "csv"},
    "jsonl": {"generator": jsonl_generator, "media_type": "application/x-ndjson", "extension": "jsonl"},
}


@app.get("/tasks/export")
def export_tasks(format: Literal["csv", "jsonl"] = Query(default="csv")):
    config = FORMAT_CONFIG[format]
    return StreamingResponse(
        content=config["generator"](tasks),
        media_type=config["media_type"],
        headers={
            "Content-Disposition": f"attachment; filename=tasks.{config['extension']}",
        },
    )

curl "http://127.0.0.1:8000/tasks/export?format=csv"
# → CSV format

curl "http://127.0.0.1:8000/tasks/export?format=jsonl"
# → {"id": 1, "title": "Setup CI/CD", ...}
#   {"id": 2, "title": "Write tests", ...}

Exercise 5: Export with a count in a header (Hard)

Create a CSV export endpoint that, on top of the streaming, sets an X-Export-Count header with the total number of records exported. The challenge: with streaming you don't know the total until you're done. The fix: count the records before you start streaming (counting is cheap, rendering is what's expensive).

from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse
from typing import Optional

app = FastAPI()

orders = [
    {"id": i, "product": f"Product {i}", "total": round(i * 15.5, 2), "status": "completed" if i % 3 == 0 else "pending"}
    for i in range(1, 51)
]


def orders_csv_generator(data: list[dict]):
    yield "id,product,total,status\n"
    for order in data:
        yield f"{order['id']},{order['product']},{order['total']},{order['status']}\n"


@app.get("/orders/export")
def export_orders(status: Optional[str] = None):
    data = orders
    if status:
        data = [o for o in orders if o["status"] == status]

    count = len(data)

    return StreamingResponse(
        content=orders_csv_generator(data),
        media_type="text/csv",
        headers={
            "Content-Disposition": "attachment; filename=orders.csv",
            "X-Export-Count": str(count),
        },
    )

curl -v "http://127.0.0.1:8000/orders/export?status=completed" 2>&1 | head -20
# The headers include: X-Export-Count: 16

Summary

• StreamingResponse sends data chunk by chunk using generators — constant memory no matter the size
• Generator functions with yield produce data incrementally — every yield sends a chunk to the client
• Content-Disposition: attachment; filename=... forces a download in the browser instead of displaying the content
• FileResponse serves existing files from disk, handling large files efficiently
• Security: always validate that file paths stay inside the allowed directory (is_relative_to)
• media_type defines the MIME type — text/csv, text/plain, application/octet-stream, etc.
• Async generators (async def + yield) enable streaming from asynchronous data sources
• StreamingResponse vs FileResponse: you generate the data → streaming; the file is on disk → file response
• Robust CSV: use Python's csv module to handle quotes, commas and special characters
• Generators get exhausted: create a new one per request — don't reuse a generator between requests

Additional resources

1. FastAPI - Custom Response — StreamingResponse, FileResponse and other response types
2. FastAPI - StreamingResponse — The specific StreamingResponse documentation
3. Python - Generators — The official Python generators tutorial
4. Python - csv module — The standard library's csv module
5. MDN - Content-Disposition — The Content-Disposition header specification
6. Starlette - Responses — Starlette's documentation (FastAPI's foundation) on responses

Next capsule: Custom Responses and Headers — You'll learn JSONResponse with custom headers, RedirectResponse, HTMLResponse, ORJSONResponse, cookies, and how to document multiple responses in OpenAPI.