Responses & Streaming

FastAPI provides several ways to manage API outputs, ranging from returning simple dictionaries to streaming large files or real-time events. While FastAPI handles most serialization automatically, you can use specific response classes and background tasks to control the response lifecycle.

Custom Response Classes

When you need to return specific content types or set custom headers and status codes, you can return a Response object directly or use a specialized subclass from fastapi.responses.

Returning Files

To efficiently stream a file from the filesystem, use FileResponse. It handles the Content-Length, Last-Modified, and ETag headers automatically.

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()

@app.get("/download")
async def download_file():
    some_file_path = "large-report.pdf"
    return FileResponse(some_file_path)

Deprecated JSON Responses

FastAPI previously recommended UJSONResponse and ORJSONResponse for performance. These are now deprecated in fastapi.responses because FastAPI now serializes Pydantic models directly to JSON bytes, which is faster and does not require a custom response class.

Streaming Responses

For large data sets or binary streams (like video or logs) that shouldn't be loaded into memory all at once, use StreamingResponse with a generator.

import anyio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"
        await anyio.sleep(0.1)

@app.get("/video")
async def main():
    return StreamingResponse(fake_video_streamer())

Server-Sent Events (SSE)

FastAPI supports Server-Sent Events (SSE) via EventSourceResponse and the ServerSentEvent model. This allows you to push real-time updates to clients over a single HTTP connection.

To implement SSE, set response_class=EventSourceResponse in your path operation and yield instances of ServerSentEvent.

from collections.abc import AsyncIterable
from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

items = [
    Item(name="Plumbus", price=32.99),
    Item(name="Portal Gun", price=999.99),
]

@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items() -> AsyncIterable[ServerSentEvent]:
    # Send a comment as a keep-alive or metadata
    yield ServerSentEvent(comment="stream of item updates")
    
    for i, item in enumerate(items):
        # data is automatically JSON-encoded
        yield ServerSentEvent(
            data=item, 
            event="item_update", 
            id=str(i + 1), 
            retry=5000
        )

Data vs Raw Data

The ServerSentEvent model provides two ways to send the payload:

• data: Accepts any JSON-serializable object (dict, Pydantic model, list, etc.). FastAPI always serializes this to JSON.
• raw_data: Accepts a str and sends it exactly as-is. Use this for pre-formatted text or non-JSON payloads.

These fields are mutually exclusive.

Background Tasks

If you need to perform logic after the response has been sent (e.g., sending an email or processing a log), use BackgroundTasks. This ensures the client doesn't have to wait for the task to finish.

In Path Operations

Inject BackgroundTasks as a parameter in your path operation function.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=""):
    with open("log.txt", mode="a") as log:
        log.write(f"notification for {email}: {message}\n")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

In Dependencies

You can also inject BackgroundTasks into dependencies. FastAPI will collect all tasks from both the dependencies and the path operation and execute them after the response is sent.

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)

def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        background_tasks.add_task(write_log, f"found query: {q}\n")
    return q

@app.post("/items/")
async def create_item(
    background_tasks: BackgroundTasks, 
    q: str = Depends(get_query)
):
    background_tasks.add_task(write_log, "item created\n")
    return {"message": "Item created"}

Important Considerations

• SSE IDs: The id field in ServerSentEvent must not contain null characters (\0).
• Dependency Cleanup: If a dependency uses yield (like a database session), FastAPI ensures the session is closed only after a StreamingResponse or EventSourceResponse has finished sending all data.
• Task Execution: Background tasks are executed after the response is sent. If the server is shut down immediately after sending the response, tasks might not complete.