FastAPI 路由处理

路由是 Web 框架的核心概念，决定了如何将 URL 路径映射到处理函数。FastAPI 提供了简洁直观的路由定义方式。

路由基础

创建路由

from fastapi import FastAPI

app = FastAPI()

# GET 请求
@app.get("/")
async def root():
    return {"message": "Hello World"}

# POST 请求
@app.post("/items/")
async def create_item():
    return {"message": "Item created"}

# PUT 请求
@app.put("/items/{item_id}")
async def update_item(item_id: int):
    return {"item_id": item_id}

# DELETE 请求
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"deleted": item_id}

HTTP 方法

FastAPI 支持所有标准 HTTP 方法：

方法	装饰器	用途
GET	@app.get()	获取资源
POST	@app.post()	创建资源
PUT	@app.put()	完整更新资源
PATCH	@app.patch()	部分更新资源
DELETE	@app.delete()	删除资源
OPTIONS	@app.options()	获取支持的方法
HEAD	@app.head()	获取响应头
TRACE	@app.trace()	诊断用

# 一个资源支持多种方法
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

@app.put("/users/{user_id}")
async def update_user(user_id: int, user: UserUpdate):
    return {"user_id": user_id, "updated": True}

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    return {"deleted": user_id}

路径参数

基本路径参数

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # item_id 会自动转换为 int 类型
    return {"item_id": item_id}

路径参数类型

# 整数
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 字符串（默认）
@app.get("/users/{username}")
async def read_user(username: str):
    return {"username": username}

# 浮点数
@app.get("/price/{amount}")
async def read_price(amount: float):
    return {"amount": amount}

# 路径（包含斜杠）
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

路径参数验证

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(
        ...,  # 必需参数
        title="商品ID",
        description="商品的唯一标识符",
        ge=1,  # 大于等于 1
        le=1000  # 小于等于 1000
    )
):
    return {"item_id": item_id}

# 多个验证条件
@app.get("/users/{user_id}")
async def read_user(
    user_id: int = Path(
        ...,
        title="用户ID",
        gt=0,      # 大于 0
        le=10000,  # 小于等于 10000
        example=123
    )
):
    return {"user_id": user_id}

数值验证参数

参数	含义
gt	大于
ge	大于等于
lt	小于
le	小于等于
multiple_of	是某数的倍数

查询参数

基本查询参数

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    # /items/?skip=0&limit=10
    return {"skip": skip, "limit": limit}

必需与可选查询参数

from typing import Optional

@app.get("/items/{item_id}")
async def read_item(
    item_id: int,
    # 必需参数（无默认值）
    name: str,
    # 可选参数（有默认值 None）
    q: Optional[str] = None,
    # 有默认值的参数
    short: bool = False
):
    item = {"item_id": item_id, "name": name}
    if q:
        item["q"] = q
    if not short:
        item["description"] = "详细描述"
    return item

查询参数验证

from fastapi import Query

@app.get("/items/")
async def read_items(
    q: str = Query(
        None,  # 默认值
        title="搜索关键词",
        description="用于搜索商品的关键词",
        min_length=3,
        max_length=50,
        regex="^[a-zA-Z0-9]+$"
    )
):
    return {"q": q}

# 列表查询参数
@app.get("/items/")
async def read_items(q: list[str] = Query(None)):
    # /items/?q=a&q=b&q=c
    return {"q": q}

路由组织

路由器（Router）

使用 APIRouter 组织路由：

# routers/users.py
from fastapi import APIRouter

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}}
)

@router.get("/")
async def list_users():
    return [{"username": "user1"}, {"username": "user2"}]

@router.get("/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

@router.post("/")
async def create_user(user: UserCreate):
    return {"username": user.username}

注册路由器

# main.py
from fastapi import FastAPI
from routers import users, items

app = FastAPI()

# 包含路由器
app.include_router(users.router)
app.include_router(items.router, prefix="/api")

# 也可以添加前缀和标签
app.include_router(
    users.router,
    prefix="/api/v1",
    tags=["v1", "users"]
)

路由组织结构

app/
├── main.py
├── routers/
│   ├── __init__.py
│   ├── users.py
│   ├── items.py
│   └── auth.py
└── models/
    └── schemas.py

路由分组与标签

from fastapi import FastAPI, APIRouter

app = FastAPI()

# 用户相关路由
users_router = APIRouter(prefix="/users", tags=["users"])

@users_router.get("/")
async def list_users():
    return []

@users_router.post("/")
async def create_user():
    return {}

# 商品相关路由
items_router = APIRouter(prefix="/items", tags=["items"])

@items_router.get("/")
async def list_items():
    return []

# 注册路由
app.include_router(users_router)
app.include_router(items_router)

路由优先级

路由按定义顺序匹配，更具体的路由应放在前面：

# 正确顺序
@app.get("/users/me")
async def read_me():
    return {"user": "current user"}

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

# 错误顺序：/users/me 会匹配到 {user_id}
@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

@app.get("/users/me")  # 永远不会匹配
async def read_me():
    return {"user": "current user"}

响应配置

响应模型

from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
    id: int
    name: str
    price: float
    description: Optional[str] = None

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    return item

# 响应模型会自动过滤未定义的字段
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    # 即使返回了额外字段，也会被过滤
    return {
        "id": item_id,
        "name": "Item Name",
        "price": 99.9,
        "description": "...",
        "secret": "hidden"  # 不会出现在响应中
    }

响应状态码

from fastapi import status

@app.post("/items/", status_code=201)
async def create_item(item: Item):
    return item

@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    return None

# 使用 status 模块
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

响应头

from fastapi import Response

@app.get("/items/")
async def read_items(response: Response):
    response.headers["X-Custom-Header"] = "Custom Value"
    response.headers["Cache-Control"] = "no-cache"
    return {"items": []}

路由装饰器选项

@app.get(
    "/items/{item_id}",
    summary="获取商品",
    description="根据 ID 获取商品详情",
    response_description="商品详情",
    deprecated=False,
    response_model=Item,
    status_code=200,
    responses={
        404: {"description": "商品不存在"},
        400: {"description": "无效的 ID"}
    }
)
async def read_item(item_id: int):
    """
    这个函数的文档字符串也会被添加到 OpenAPI 文档中。
    
    支持 Markdown 格式：
    - 列表项1
    - 列表项2
    """
    return {"item_id": item_id}

静态文件路由

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 挂载静态文件目录
app.mount("/static", StaticFiles(directory="static"), name="static")

# 访问 /static/images/logo.png

小结

本章我们学习了：

1. 路由的基本定义方式
2. 路径参数和查询参数的使用
3. 参数验证和类型转换
4. 使用 APIRouter 组织路由
5. 响应模型和状态码配置

练习

1. 创建一个 CRUD 路由组（增删改查）
2. 为路由添加参数验证
3. 使用 APIRouter 组织多个路由模块
4. 配置自定义响应模型