FastAPI 请求对象（Request）和 响应类型（Response）

在 FastAPI 中，请求对象（Request） 和 响应类型（Response） 的设计非常灵活且现代化。

✅ 一、请求对象（Request）

通常你不需要显式使用 Request 对象，因为 FastAPI 已自动解析路径、查询、请求体。

但以下场景需要用到：

🔹 场景 1：获取客户端 IP、User-Agent、Headers 等元信息

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/info")
def get_client_info(request: Request):
    return {
        "client_host": request.client.host,
        "user_agent": request.headers.get("user-agent"),
        "url": str(request.url),
    }

✅ 适用于日志、审计、限流等中间件逻辑。

🔹 场景 2：处理原始 body（如 Webhook、非 JSON 数据）

@app.post("/webhook")
async def handle_webhook(request: Request):
    raw_body = await request.body()  # bytes
    # 处理 GitHub/GitLab webhook 等
    return {"received": len(raw_body)}

⚠️ 注意：如果用了 request.body()，就不能再用 Pydantic 模型接收 JSON！

✅ 二、响应类型详解

FastAPI 默认将返回值转为 JSON 响应（application/json），但支持多种响应类型。

🔸 1. 普通 JSON 响应（最常用）

from pydantic import BaseModel

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

@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int):
    return {"id": user_id, "name": "Alice", "email": "alice@example.com"}

• ✅ 使用 response_model 明确指定返回结构
• ✅ 自动过滤模型中未定义的字段（安全！）
• ✅ 自动生成 OpenAPI 文档

🔸 2. 列表响应（List of Items）

@app.get("/users/", response_model=list[UserOut])
def list_users():
    return [
        {"id": 1, "name": "Alice", "email": "a@example.com"},
        {"td": 2, "name": "Bob", "email": "b@example.com"},
    ]

💡 Python 3.9+ 直接用 list[UserOut]；旧版用 List[UserOut]（需 from typing import List）

🔸 3. 分页列表（Pagination）—— 主流写法

这是 API 设计的最佳实践！推荐返回 结构化分页对象：

from pydantic import BaseModel
from typing import Generic, TypeVar

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    size: int
    pages: int

@app.get("/users/", response_model=PaginatedResponse[UserOut])
def list_users_paginated(page: int = 1, size: int = 10):
    # 模拟数据库查询
    total = 125
    items = [
        {"id": i, "name": f"User{i}", "email": f"user{i}@example.com"}
        for i in range((page-1)*size, min(page*size, total))
    ]
    return PaginatedResponse(
        items=items,
        total=total,
        page=page,
        size=size,
        pages=(total + size - 1) // size
    )

响应示例：

{
  "items": [{"id":1,"name":"User1",...}, ...],
  "total": 125,
  "page": 1,
  "size": 10,
  "pages": 13
}

✅ 优点：前端可轻松实现分页 UI，无需猜测总数或页数。

🔸 4. 文件上传（Form Data / Multipart）

from fastapi import File, UploadFile
import shutil

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    # 保存文件
    with open(f"uploads/{file.filename}", "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"filename": file.filename, "content_type": file.content_type}

• ✅ 使用 UploadFile 支持大文件（流式读取）
• ✅ 自动校验 multipart/form-data

🔸 5. 文件下载（StreamingResponse / FileResponse）

方式 A：小文件 → FileResponse

from fastapi.responses import FileResponse

@app.get("/download/report.pdf")
def download_report():
    return FileResponse(
        path="reports/report.pdf",
        filename="monthly_report.pdf",  # 下载时的文件名
        media_type="application/pdf"
    )

方式 B：大文件 / 动态生成 → StreamingResponse

from fastapi.responses import StreamingResponse
import io

@app.get("/export/csv")
def export_csv():
    def generate():
        yield "name,age\n"
        yield "Alice,30\n"
        yield "Bob,25\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=users.csv"}
    )

🔸 6. 其他响应类型

示例：创建资源后返回 201 + Location 头

from fastapi import status
from fastapi.responses import JSONResponse

@app.post("/users/", status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate):
    user_id = 123  # 模拟 DB 插入
    return JSONResponse(
        content={"id": user_id, "name": user.name},
        headers={"Location": f"/users/{user_id}"}
    )