FastAPI 后端开发实战笔记

环境配置

• Python 版本：Python 3.12+（建议 3.10 以上）
• 开发工具：PyCharm 或 VS Code
• 操作系统：Windows / macOS / Linux

为什么我会选 FastAPI

在 Python 生态里，FastAPI 之所以常被拿来做 API 服务，主要是这几件事比较顺手：速度够快，类型提示能直接参与参数校验和文档生成，异步支持也做得比较自然。对写接口的人来说，少写不少样板代码。

这套东西更适合前后端分离、接口先行的项目。要是只是写几个小脚本式接口，当然不必上来就追求'最完整'的架构；但一旦接口数量开始上升，FastAPI 的结构感就会更明显。

环境搭建

先装 FastAPI 和 Web 服务器 uvicorn：

pip install fastapi uvicorn

ASGI 是怎么回事

ASGI（Asynchronous Server Gateway Interface）是 Python 异步 Web 服务和应用之间的标准接口。和老一些的 WSGI 比起来，它把异步请求、WebSocket 这些场景也纳进来了。

协议	特性	代表框架
WSGI	同步、单线程	Flask、Django 早期
ASGI	异步、支持 WebSocket	FastAPI、Django Channels

ASGI 应用本质上是一个可调用对象，接收 scope、receive、send 三个参数：

async def application(scope, receive, send):
    # scope: 包含请求信息（HTTP 方法、路径、Headers 等）
    # receive: 异步函数，用于接收请求体
    # send: 异步函数，用于发送响应
    await send({'type': 'http.response.start', 'status': 200, 'headers': [(b'content-type', b'text/plain')]})
    await send({'type': 'http.response.body', 'body': b'Hello, ASGI!'})

FastAPI 本身就是 ASGI 应用框架，底层依赖 Starlette。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
    }

验证大致会走这几步：类型检查、约束检查、默认值填充，然后才是自定义验证器。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)

特性	中间件	依赖注入
执行时机	所有请求	特定路由
粒度	全局	局部
使用场景	日志、CORS、压缩	数据库连接、认证
返回值	修改 request/response	注入到路由函数

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)
    }