Skip to main content

Response Handling

Response Models​

Basic Response Model​

from pydantic import BaseModel

class ItemResponse(BaseModel):
    name: str
    price: float
    is_offer: bool = False

@app.post("/items/", response_model=ItemResponse)
async def create_item(item: Item):
    return item

Multiple Response Models​

from typing import Union

class UserBase(BaseModel):
    username: str
    email: str

class UserCreate(UserBase):
    password: str

class UserResponse(UserBase):
    id: int
    is_active: bool

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    return user

Response Model with Excludes​

class UserInDB(BaseModel):
    username: str
    email: str
    hashed_password: str

@app.get("/users/me", response_model=UserInDB, response_model_exclude={"hashed_password"})
async def get_current_user():
    return current_user

# Or include only specific fields
@app.get("/users/me", response_model=UserInDB, response_model_include={"username", "email"})
async def get_current_user():
    return current_user

Status Codes​

Custom Status Codes​

from fastapi import status

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    return None

# Common status codes
@app.get("/items/", status_code=200)  # HTTP_200_OK
@app.post("/items/", status_code=201)  # HTTP_201_CREATED
@app.put("/items/{item_id}", status_code=200)  # HTTP_200_OK
@app.delete("/items/{item_id}", status_code=204)  # HTTP_204_NO_CONTENT

Dynamic Status Codes​

from fastapi import Response

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, response: Response):
    if item_id in items:
        response.status_code = status.HTTP_200_OK
        return {"message": "Item updated"}
    else:
        response.status_code = status.HTTP_201_CREATED
        return {"message": "Item created"}

Response Headers​

Custom Headers​

from fastapi import Response

@app.get("/items/{item_id}")
async def read_item(item_id: int, response: Response):
    response.headers["X-Cat-Dog"] = "alone in the world"
    return {"item_id": item_id}

# Multiple headers
@app.get("/items/{item_id}")
async def read_item(item_id: int, response: Response):
    response.headers.update({
        "X-Custom-Header": "value",
        "X-Another-Header": "another-value"
    })
    return {"item_id": item_id}

Response Headers in Dependencies​

from fastapi import Depends, Header

async def common_parameters(
    response: Response,
    x_token: str = Header(...)
):
    response.headers["X-Token-Used"] = x_token
    return {"token": x_token}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

Response Types​

JSON Response​

from fastapi.responses import JSONResponse

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    return JSONResponse(
        content={"item_id": item_id, "message": "Hello World"},
        status_code=200,
        headers={"X-Custom-Header": "value"}
    )

HTML Response​

from fastapi.responses import HTMLResponse

@app.get("/items/{item_id}", response_class=HTMLResponse)
async def read_item(item_id: str):
    return f"""
    <html>
        <head>
            <title>Item {item_id}</title>
        </head>
        <body>
            <h1>Item: {item_id}</h1>
        </body>
    </html>
    """

Plain Text Response​

from fastapi.responses import PlainTextResponse

@app.get("/items/{item_id}", response_class=PlainTextResponse)
async def read_item(item_id: str):
    return f"Item ID: {item_id}"

File Response​

from fastapi.responses import FileResponse

@app.get("/download/{filename}")
async def download_file(filename: str):
    return FileResponse(
        path=f"files/{filename}",
        filename=filename,
        media_type='application/octet-stream'
    )

# With custom headers
@app.get("/download/{filename}")
async def download_file(filename: str):
    return FileResponse(
        path=f"files/{filename}",
        filename=filename,
        headers={"X-Custom-Header": "value"}
    )

Streaming Response​

from fastapi.responses import StreamingResponse
import io

@app.get("/generate-csv")
async def generate_csv():
    def generate():
        yield "id,name,email\n"
        for i in range(1000):
            yield f"{i},User {i},user{i}@example.com\n"

    return StreamingResponse(
        generate(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=users.csv"}
    )

# Stream file content
@app.get("/stream-file/{filename}")
async def stream_file(filename: str):
    file_path = f"files/{filename}"

    def iterfile():
        with open(file_path, mode="rb") as file_like:
            yield from file_like

    return StreamingResponse(iterfile(), media_type="application/octet-stream")

Response Cookies​

Setting Cookies​

from fastapi import Response

@app.post("/login/")
async def login(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123",
        max_age=3600,
        httponly=True,
        secure=True,
        samesite="lax"
    )
    return {"message": "Login successful"}

# Multiple cookies
@app.post("/login/")
async def login(response: Response):
    response.set_cookie("session_id", "abc123", httponly=True)
    response.set_cookie("user_preference", "dark_mode", max_age=86400)
    return {"message": "Login successful"}

Deleting Cookies​

@app.post("/logout/")
async def logout(response: Response):
    response.delete_cookie("session_id")
    return {"message": "Logout successful"}

Error Responses​

HTTP Exceptions​

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "Item not found"}
        )
    return {"item": items[item_id]}

Custom Exception Handlers​

from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something wrong"}
    )

# Override default HTTP exception handler
@app.exception_handler(404)
async def not_found_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=404,
        content={"message": "Resource not found", "url": str(request.url)}
    )

Validation Error Responses​

from fastapi import Request, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"message": "Validation error", "errors": exc.errors()}
    )

Response Middleware​

Adding Response Headers​

@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    return response

Response Timing​

import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Response Caching​

Cache Headers​

from fastapi import Response

@app.get("/static-data")
async def get_static_data(response: Response):
    response.headers["Cache-Control"] = "public, max-age=3600"
    response.headers["ETag"] = "abc123"
    return {"data": "This is static data"}

# No cache
@app.get("/dynamic-data")
async def get_dynamic_data(response: Response):
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    response.headers["Pragma"] = "no-cache"
    response.headers["Expires"] = "0"
    return {"data": "This is dynamic data"}

Conditional Responses​

from fastapi import Header

@app.get("/items/{item_id}")
async def read_item(
    item_id: str,
    if_none_match: str = Header(None),
    response: Response
):
    item = get_item(item_id)
    current_etag = generate_etag(item)

    if if_none_match == current_etag:
        response.status_code = 304
        return None

    response.headers["ETag"] = current_etag
    return item