响应处理与状态码

一、响应模型（Response Model）

基本用法

使用 response_model 定义响应数据结构：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    email: str

class UserCreate(BaseModel):
    name: str
    email: str
    password: str

@app.post("/users/", response_model=User)
async def create_user(user: UserCreate):
    # 返回时自动过滤掉 password 字段
    return {"id": 1, **user.model_dump()}

响应模型的作用

  1. 过滤字段：自动排除未定义的字段
  2. 类型转换：确保输出格式正确
  3. 文档生成：自动更新 Swagger 文档

排除字段

@app.post("/users/", response_model=User, response_model_exclude={"id"})
async def create_user(user: UserCreate):
    return {"id": 1, **user.model_dump()}

@app.get("/users/{user_id}", response_model=User, response_model_exclude_unset=True)
async def get_user(user_id: int):
    # 只返回已设置的字段
    return {"id": user_id, "name": "张三", "email": "test@example.com"}

包含/排除嵌套字段

@app.get("/items/{item_id}", response_model=Item, response_model_include={"name", "price"})
async def get_item(item_id: int):
    return item

@app.get("/users/{user_id}", response_model=User, response_model_exclude={"password", "secret_field"})
async def get_user(user_id: int):
    return user

二、状态码（Status Code）

设置状态码

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=201)
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

常用状态码

状态码常量说明
200HTTP_200_OK成功（默认）
201HTTP_201_CREATED创建成功
204HTTP_204_NO_CONTENT无内容（删除成功）
400HTTP_400_BAD_REQUEST请求错误
401HTTP_401_UNAUTHORIZED未授权
403HTTP_403_FORBIDDEN禁止访问
404HTTP_404_NOT_FOUND资源不存在
422HTTP_422_UNPROCESSABLE_ENTITY验证失败
500HTTP_500_INTERNAL_SERVER_ERROR服务器错误

三、Response 对象

使用 JSONResponse

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id > 100:
        return JSONResponse(
            status_code=404,
            content={"message": "Item not found"}
        )
    return {"item_id": item_id}

设置响应头

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/items/")
async def read_items():
    return JSONResponse(
        content={"items": []},
        headers={"X-Custom-Header": "custom-value"}
    )
from fastapi import FastAPI, Response

app = FastAPI()

@app.post("/login/")
async def login(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123",
        httponly=True,
        max_age=1800
    )
    return {"message": "Logged in"}

四、不同响应类型

HTMLResponse

from fastapi.responses import HTMLResponse

@app.get("/html/", response_class=HTMLResponse)
async def get_html():
    return """
    <html>
        <head><title>Hello</title></head>
        <body><h1>Hello, FastAPI!</h1></body>
    </html>
    """

PlainTextResponse

from fastapi.responses import PlainTextResponse

@app.get("/text/", response_class=PlainTextResponse)
async def get_text():
    return "Hello, FastAPI!"

FileResponse

from fastapi.responses import FileResponse

@app.get("/download/")
async def download_file():
    return FileResponse(
        path="files/report.pdf",
        filename="report.pdf",
        media_type="application/pdf"
    )

StreamingResponse

from fastapi.responses import StreamingResponse
import io

@app.get("/stream/")
async def stream_data():
    async def generate():
        for i in range(10):
            yield f"data: {i}\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/plain"
    )

RedirectResponse

from fastapi.responses import RedirectResponse

@app.get("/redirect/")
async def redirect_to_docs():
    return RedirectResponse(url="/docs")

五、异常处理

HTTPException

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id > 100:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "Not Found"}
        )
    return {"item_id": item_id}

自定义异常

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

class ItemNotFoundException(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

app = FastAPI()

@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request: Request, exc: ItemNotFoundException):
    return JSONResponse(
        status_code=404,
        content={"message": f"Item {exc.item_id} not found"}
    )

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id > 100:
        raise ItemNotFoundException(item_id)
    return {"item_id": item_id}

全局异常处理

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "message": "验证失败"}
    )

@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
    return JSONResponse(
        status_code=500,
        content={"message": "服务器内部错误", "detail": str(exc)}
    )

六、自定义响应模型

多种响应模型

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

class SuccessResponse(BaseModel):
    success: bool = True
    data: dict

class ErrorResponse(BaseModel):
    success: bool = False
    message: str
    code: int

@app.get("/items/{item_id}", responses={
    200: {"model": SuccessResponse, "description": "成功"},
    404: {"model": ErrorResponse, "description": "未找到"}
})
async def read_item(item_id: int):
    if item_id > 100:
        return JSONResponse(
            status_code=404,
            content={"success": False, "message": "Item not found", "code": 404}
        )
    return {"success": True, "data": {"item_id": item_id}}

七、分页响应

标准分页模型

from pydantic import BaseModel
from typing import Generic, TypeVar, List

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    page: int
    page_size: int
    total_pages: int

@app.get("/items/", response_model=PaginatedResponse[Item])
async def list_items(page: int = 1, page_size: int = 10):
    items = get_items_from_db(skip=(page-1)*page_size, limit=page_size)
    total = count_items_in_db()
    return {
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "total_pages": (total + page_size - 1) // page_size
    }

八、完整示例

from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel
from typing import Generic, TypeVar, List

app = FastAPI()

# 模型定义
T = TypeVar("T")

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

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

class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    data: T | None = None
    message: str = ""

class PaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    page: int
    page_size: int

# 模拟数据库
items_db = {}
item_counter = 0

# API 端点
@app.post("/items/", 
    response_model=ApiResponse[Item],
    status_code=status.HTTP_201_CREATED
)
async def create_item(item: ItemCreate):
    global item_counter
    item_counter += 1
    new_item = Item(id=item_counter, **item.model_dump())
    items_db[item_counter] = new_item
    return ApiResponse(data=new_item, message="创建成功")

@app.get("/items/{item_id}", response_model=ApiResponse[Item])
async def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return ApiResponse(data=items_db[item_id])

@app.get("/items/", response_model=PaginatedResponse[Item])
async def list_items(page: int = 1, page_size: int = 10):
    all_items = list(items_db.values())
    start = (page - 1) * page_size
    end = start + page_size
    return {
        "items": all_items[start:end],
        "total": len(all_items),
        "page": page,
        "page_size": page_size
    }

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    del items_db[item_id]
    return None

九、小结

功能用法
响应模型response_model=Model
状态码status_code=201 或 status.HTTP_201_CREATED
排除字段response_model_exclude={"field"}
JSONResponse自定义状态码和响应头
HTTPExceptionraise HTTPException(status_code=404, detail="...")
自定义异常@app.exception_handler(CustomException)