FastAPI 后端开发实战笔记

Hello World

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "你好，全栈之路！"}

# 启动命令：uvicorn main:app --reload

--reload 在开发时很好用，代码一保存就会自动重启。线上别开，没必要。

Pydantic 数据验证

FastAPI 和 Pydantic 搭在一起，参数校验基本不用自己手写。

from pydantic import BaseModel

# 定义一个商品的数据结构
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool | None = None

@app.post("/items/")
def create_item(item: Item):
    # 如果用户传的 price 不是数字，FastAPI 会自动返回 422 错误
    return {
        "item_name": item.name,
        "total_price": item.price * 1.2
    }

验证大致会走这几步：类型检查、约束检查、默认值填充，然后才是自定义验证器。Pydantic V2 里常见的是 @field_validator，老项目里也可能还会见到 @validator。

自定义验证器示例

from pydantic import BaseModel, field_validator

class User(BaseModel):
    username: str
    age: int
    email: str

    @field_validator('username')
    @classmethod
    def validate_username(cls, v):
        if len(v) < 3:
            raise ValueError('用户名至少 3 个字符')
        if not v.isalnum():
            raise ValueError('用户名只能包含字母和数字')
        return v

    @field_validator('age')
    @classmethod
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError('年龄必须在 0-150 之间')
        return v

app = FastAPI()

@app.post("/users/")
def create_user(user: User):
    return {"message": "用户创建成功", "user": user}

嵌套模型验证

from typing import List
from pydantic import BaseModel

class Address(BaseModel):
    city: str
    street: str
    zipcode: str

class UserWithAddress(BaseModel):
    name: str
    addresses: List[Address]

@app.post("/users-with-address/")
def create_user_with_address(user: UserWithAddress):
    return user

路径参数和查询参数

路径参数通常用来标识资源，比如 /items/42；查询参数更常见于搜索和分页，比如 /items/?skip=0&limit=10。

@app.get("/users/{user_id}")
def read_user(user_id: int, q: str | None = None):
    # user_id 会自动转成整数，q 是可选的搜索关键词
    return {"user_id": user_id, "query": q}

依赖注入

FastAPI 的 Depends 用起来不复杂，但很实用。它把'怎么拿到这个东西'从路由函数里拆出去，尤其适合数据库连接、用户鉴权、公共参数这类逻辑。

from fastapi import Depends, FastAPI

app = FastAPI()

# 定义一个依赖函数
def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 使用依赖
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/users/")
def read_users(commons: dict = Depends(common_parameters)):
    return commons

同一个请求里，相同依赖默认只会执行一次，这点对有缓存或连接对象的场景挺省事。

类作为依赖

from fastapi import Depends, FastAPI

app = FastAPI()

class DatabaseConnection:
    def __init__(self):
        self.connection = "模拟数据库连接"
    
    def query(self, sql: str):
        return f"执行：{sql}"
    
    def close(self):
        self.connection = None

def get_db():
    db = DatabaseConnection()
    try:
        yield db
    finally:
        db.close()

@app.get("/data/")
def get_data(db: DatabaseConnection = Depends(get_db)):
    result = db.query("SELECT * FROM users")
    return {"result": result}

子依赖与依赖链

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()

# 基础依赖：验证 Token
def verify_token(x_token: str = Header(...)):
    if x_token != "secret-token":
        raise HTTPException(status_code=400, detail="Token 无效")
    return x_token

# 子依赖：获取当前用户（依赖 verify_token）
def get_current_user(token: str = Depends(verify_token)):
    return {"username": "zhangsan", "id": 1}

@app.get("/profile/")
def read_profile(user: dict = Depends(get_current_user)):
    return user

中间件

中间件放在路由前后都能插一层逻辑，常见用途是日志、跨域、压缩、统一异常处理。它和依赖注入的区别很明显：中间件偏全局，依赖偏局部。

请求执行顺序通常是这样的：

请求 -> 中间件 1 -> 中间件 2 -> 路由处理 -> 中间件 2 -> 中间件 1 -> 响应

自定义中间件

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

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    print(f"[请求] {request.method}{request.url.path}")
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    print(f"[响应] 耗时：{process_time:.4f}s")
    return response

@app.middleware("http")
async def catch_exceptions(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})

@app.get("/")
def read_root():
    return {"message": "Hello"}

内置中间件

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

# CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

# GZip 压缩中间件
app.add_middleware(GZipMiddleware, minimum_size=1000)

Web 安全基础

CSRF（跨站请求伪造）

攻击方式很直接：诱导用户在已经登录的网站上发起原本不想执行的操作。它不是 FastAPI 独有的问题，前后端只要有 Cookie 会话就得考虑。

防护措施：

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

app = FastAPI()

async def verify_csrf_token(request: Request):
    csrf_token = request.headers.get("X-CSRF-Token")
    stored_token = request.cookies.get("csrf_token")
    if not csrf_token or csrf_token != stored_token:
        raise HTTPException(status_code=403, detail="CSRF Token 无效")
    return True

@app.post("/transfer/")
async def transfer_money(request: Request):
    await verify_csrf_token(request)
    return {"message": "转账成功"}

最佳实践：

• 使用 SameSite Cookie 属性。
• 验证 Referer/Origin Header。
• 关键操作加二次确认。

XSS（跨站脚本攻击）

攻击者把恶意脚本塞进页面输入里，浏览器一旦直接渲染，Cookie、会话信息或页面行为都可能被牵着走。

防护措施：

from fastapi import FastAPI
from pydantic import BaseModel, field_validator
import html

app = FastAPI()

class Comment(BaseModel):
    content: str

    @field_validator('content')
    @classmethod
    def sanitize_content(cls, v):
        return html.escape(v)

@app.post("/comments/")
def create_comment(comment: Comment):
    return {"content": comment.content}

最佳实践：

• 对所有用户输入进行转义。
• 使用 Content Security Policy (CSP)。
• 设置 HttpOnly Cookie。

SQL 注入防护

这类问题的根源还是把用户输入直接拼进 SQL。只要用了参数化查询，很多低级坑就会少一大半。

防护措施：

from fastapi import FastAPI, HTTPException
import sqlite3

app = FastAPI()

# 正确做法：使用参数化查询
@app.get("/users/safe/{user_id}")
def get_user_safe(user_id: int):
    conn = sqlite3.connect("test.db")
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
    return cursor.fetchone()

最佳实践：

• 永远使用参数化查询/预编译语句。
• ORM（如 SQLAlchemy）自动处理转义。
• 保持最小权限。

安全头部配置

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

@app.middleware("http")
async def add_security_headers(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"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    return response

app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"])

FastAPI 0.100+ 的几个变化

Pydantic V2 支持

FastAPI 0.100+ 默认跟着 Pydantic V2 走，模型定义更严格，也更适合做约束清晰的接口。

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0, description="必须大于 0")
    model_config = {
        "strict": True,
        "extra": "forbid"
    }

Annotated 语法

from typing import Annotated
from fastapi import FastAPI, Query, Path

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: Annotated[int, Path(title="项目 ID", ge=1)],
    q: Annotated[str | None, Query(min_length=3)] = None,
    limit: Annotated[int, Query(le=100)] = 10
):
    return {"item_id": item_id, "q": q, "limit": limit}

改进异常处理

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

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": " -> ".join(str(x) for x in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })
    return JSONResponse(
        status_code=422,
        content={"detail": "数据验证失败", "errors": errors}
    )

自动化文档

FastAPI 默认就会把 OpenAPI 文档和交互页面带出来。服务启动后访问：

• http://127.0.0.1:8000/docs

页面里能直接点 'Try it out' 调接口。对开发阶段来说，这比单独翻接口文档省事得多。

一个简单的待办事项 API

下面这个例子功能很少，但把增删查的骨架都串起来了。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

todo_db = []

class Todo(BaseModel):
    id: int
    title: str
    completed: bool = False

@app.post("/todos/", status_code=201)
def add_todo(todo: Todo):
    todo_db.append(todo)
    return {"message": "添加成功", "data": todo}

@app.get("/todos/")
def list_todos():
    return todo_db

@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
    for t in todo_db:
        if t.id == todo_id:
            return t
    raise HTTPException(status_code=404, detail="任务找不到了...")

@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
    global todo_db
    todo_db = [t for t in todo_db if t.id != todo_id]
    return {"message": "删除成功"}

开发时我更在意的几件事

• async def 别乱用。如果函数里没有 await，直接写 def 往往更合适。
• 状态码别省。200、201、404、401、403 这些语义清楚，前端和排障都省时间。
• Depends 很适合放登录校验和公共参数，能把路由函数写得干净一点。

练习

题目 1：查询参数与校验

编写一个 API GET /search/，接收参数 keyword（必填，长度至少 2）和 limit（选填，默认 10，最大 50）。使用 Query 进行参数校验。

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/search/")
def search(
    keyword: str = Query(..., min_length=2, description="搜索关键词"),
    limit: int = Query(10, le=50, description="返回结果数量")
):
    return {"keyword": keyword, "limit": limit, "results": ["模拟数据 1", "模拟数据 2"]}

题目 2：依赖注入与 Token 验证

编写一个依赖函数 verify_token，检查请求头 x-token 是否为 "fake-super-secret-token"。如果是，允许访问；否则抛出 400 错误。将此依赖应用到 GET /protected/ 路由上。

from fastapi import FastAPI, Header, HTTPException, Depends

app = FastAPI()

def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="Token 无效")
    return x_token

@app.get("/protected/")
def protected_route(token: str = Depends(verify_token)):
    return {"message": "验证通过", "token": token}

题目 3：文件上传

编写一个 API POST /upload/，接收一个文件上传。返回文件的文件名和文件大小。需要安装 python-multipart。

from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(content)
    }