HowToCodePage

File uploads, streaming and WebSockets

Accept uploads without running out of memory, stream large responses, serve static files, and add a WebSocket endpoint with proper disconnect handling.

Uploads

from fastapi import APIRouter, File, Form, HTTPException, UploadFile
import aiofiles, uuid, pathlib

router = APIRouter()
UPLOAD_DIR = pathlib.Path("var/uploads")
MAX_BYTES = 5 * 1024 * 1024
ALLOWED = {"image/png", "image/jpeg", "application/pdf"}

@router.post("/files")
async def upload(file: UploadFile = File(...), note: str = Form("")):
    if file.content_type not in ALLOWED:
        raise HTTPException(415, "Unsupported media type")

    dest = UPLOAD_DIR / f"{uuid.uuid4().hex}{pathlib.Path(file.filename or '').suffix}"
    size = 0
    async with aiofiles.open(dest, "wb") as out:
        while chunk := await file.read(1024 * 1024):     # 1 MiB at a time
            size += len(chunk)
            if size > MAX_BYTES:
                await out.close()
                dest.unlink(missing_ok=True)
                raise HTTPException(413, "File too large")
            await out.write(chunk)
    return {"stored_as": dest.name, "bytes": size}
  • Small uploads sit in memory; larger ones spill to a temporary file. Reading in chunks keeps your own memory use bounded regardless.
  • Do not call await file.read() with no argument on an endpoint that accepts untrusted input.
  • Enable python-multipart; without it, form parsing fails at startup with a clear error.

Streaming responses and static files

from fastapi.responses import StreamingResponse
from fastapi.staticfiles import StaticFiles

@app.get("/export.csv")
def export(db: Session = Depends(get_db)):
    def rows():
        yield "id,name\n"
        for item in db.scalars(select(Item).execution_options(yield_per=1000)):
            yield f"{item.id},{item.name}\n"
    return StreamingResponse(rows(), media_type="text/csv",
                             headers={"Content-Disposition": 'attachment; filename="items.csv"'})

app.mount("/static", StaticFiles(directory="static"), name="static")

A generator response starts sending headers immediately, so an error after the first chunk cannot become a 500. Validate everything you can before yielding the first byte.

WebSockets

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/echo")
async def echo(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            message = await ws.receive_text()
            await ws.send_text(f"echo: {message}")
    except WebSocketDisconnect:
        logger.info("client disconnected")
⚠️
Never build a filesystem path from file.filename. The value is client-controlled and can contain path separators or traversal sequences. Generate your own name, keep only a validated extension, and store the original name in the database if you need to show it.

FAQ

How do I limit upload size before reading the body?
Enforce it at the proxy (client_max_body_size in nginx) or in middleware that inspects Content-Length and rejects early. Application-level counting is a second line of defence, not the first.
Do WebSockets work behind a load balancer?
Only with upgrade support enabled. nginx needs proxy_set_header Upgrade and Connection, and the idle timeout must exceed your ping interval, or connections will be closed silently.

Background tasks and lifespan events Your first FastAPI app

Last refreshed 2026-09-18.