Python 全栈开发：FastAPI 高性能后端开发

FastAPI 与 ASGI 的关系

FastAPI 是一个 ASGI 应用框架，内部使用 Starlette 处理 ASGI 协议。运行 uvicorn main:app 时：

1. Uvicorn 作为 ASGI 服务器启动。
2. 接收到的 HTTP 请求被解析成 ASGI 事件。
3. FastAPI（通过 Starlette）处理这些事件。
4. 响应通过 ASGI 协议返回给客户端。

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
    }

验证流程

1. 类型检查：确保每个字段符合声明的类型。
2. 约束验证：检查 min_length、max_length、ge、le 等。
3. 默认值填充：为可选字段填充默认值。
4. 自定义验证器：执行 @validator 或 @field_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（指定 ID）。
• 查询参数：/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}

依赖注入机制

什么是依赖注入？

依赖注入（Dependency Injection, DI）是一种设计模式，将对象的创建和管理从使用它的代码中分离出来。

FastAPI 依赖注入的核心机制

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

依赖的执行流程

1. 解析依赖树：分析路由函数的参数，识别所有 Depends 标记的依赖。
2. 递归解析：如果依赖 A 依赖 B，先解析 B，再解析 A。
3. 缓存机制：同一个请求中，相同的依赖只执行一次。
4. 注入执行：将解析结果作为参数传给路由函数。

类作为依赖

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

中间件机制

什么是中间件？

中间件是在请求到达路由处理函数之前和响应返回客户端之前执行的代码。它可以记录日志、添加响应头、处理跨域（CORS）、认证授权、压缩响应等。

中间件执行顺序

请求 -> 中间件 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（跨站请求伪造）

攻击原理：诱导用户在已登录的网站上执行非预期的操作。

防护措施：

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 里，你什么都不用做。服务跑起来后，直接访问：

• http://127.0.0.1:8000/docs

你会看到一个超级精美的交互式文档。可以在上面直接点"Try it out"测试你的接口。

实战案例

构建一个待办事项 (To-Do) 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": "删除成功"}

开发建议

1. async def 别乱用：如果函数里没有 await 任何东西，直接写 def 往往更快。
2. 状态码很重要：成功了返回 200/201，找不到了返回 404，权限不够返回 401/403。
3. 依赖注入：处理用户登录（Token 验证）时，使用 FastAPI 的 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)
    }