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

子应用与挂载

当需要将多个独立的 FastAPI 应用组合在一起，或者需要某个路径下的应用拥有独立的文档和配置时，可以使用子应用（Sub Applications）和挂载（Mount）。

为什么需要子应用

使用 APIRouter 可以组织路由，但它不是独立的应用。子应用则是完全独立的 FastAPI 应用实例：

特性	APIRouter	子应用
独立的 OpenAPI 文档	否	是
独立的 Swagger UI	否	是
独立的中间件	否	是
独立的生命周期	否	是
共享应用状态	是	否

创建子应用

from fastapi import FastAPI

# 主应用
app = FastAPI()

# 子应用
sub_app = FastAPI(
    title="子应用 API",
    description="这是一个独立的子应用",
    docs_url="/docs",
    redoc_url="/redoc"
)

@sub_app.get("/")
async def sub_root():
    return {"message": "子应用首页"}

@sub_app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 挂载子应用
app.mount("/subapi", sub_app)

访问方式：

• 子应用首页：http://localhost:8000/subapi/
• 子应用文档：http://localhost:8000/subapi/docs
• 子应用路由：http://localhost:8000/subapi/items/1

实际应用场景

场景一：模块化大型应用

from fastapi import FastAPI
from pydantic import BaseModel

# 用户模块（独立应用）
users_app = FastAPI(title="用户模块", docs_url="/docs")

class User(BaseModel):
    id: int
    name: str

@users_app.get("/")
async def list_users():
    return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

@users_app.get("/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": f"User{user_id}"}

# 商品模块（独立应用）
products_app = FastAPI(title="商品模块", docs_url="/docs")

@products_app.get("/")
async def list_products():
    return [{"id": 1, "name": "商品A"}, {"id": 2, "name": "商品B"}]

# 主应用
app = FastAPI(title="主应用")

@app.get("/")
async def root():
    return {"message": "主应用首页"}

# 挂载子应用
app.mount("/users", users_app)
app.mount("/products", products_app)

访问：

• 主应用文档：http://localhost:8000/docs
• 用户模块文档：http://localhost:8000/users/docs
• 商品模块文档：http://localhost:8000/products/docs

场景二：独立的管理后台

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBasic, HTTPBasicCredentials

# 管理后台应用
admin_app = FastAPI(
    title="管理后台",
    docs_url="/docs",
    openapi_url="/openapi.json"
)

security = HTTPBasic()

def verify_admin(credentials: HTTPBasicCredentials = Depends(security)):
    if credentials.username != "admin" or credentials.password != "secret":
        raise HTTPException(status_code=401, detail="需要管理员权限")
    return credentials.username

@admin_app.get("/")
async def admin_home(username: str = Depends(verify_admin)):
    return {"message": f"欢迎, {username}"}

@admin_app.get("/stats")
async def admin_stats(username: str = Depends(verify_admin)):
    return {"total_users": 100, "total_items": 500}

# 主应用（公开 API）
app = FastAPI(title="公开 API")

@app.get("/")
async def root():
    return {"message": "公开首页"}

@app.get("/public")
async def public_endpoint():
    return {"data": "公开数据"}

# 挂载管理后台
app.mount("/admin", admin_app)

场景三：不同版本 API

from fastapi import FastAPI
from pydantic import BaseModel

# V1 版本
v1_app = FastAPI(title="API V1", docs_url="/docs")

class ItemV1(BaseModel):
    name: str
    price: float

@v1_app.post("/items")
async def create_item_v1(item: ItemV1):
    return {"version": "v1", "item": item}

# V2 版本
v2_app = FastAPI(title="API V2", docs_url="/docs")

class ItemV2(BaseModel):
    name: str
    price: float
    description: str | None = None
    category: str | None = None

@v2_app.post("/items")
async def create_item_v2(item: ItemV2):
    return {"version": "v2", "item": item}

# 主应用
app = FastAPI(title="API Gateway")

@app.get("/")
async def root():
    return {"versions": ["v1", "v2"]}

# 挂载不同版本
app.mount("/v1", v1_app)
app.mount("/v2", v2_app)

子应用的生命周期

重要：子应用不会执行主应用的 lifespan 事件。如果子应用需要独立的启动和关闭逻辑：

from contextlib import asynccontextmanager
from fastapi import FastAPI

# 子应用的生命周期
@asynccontextmanager
async def sub_lifespan(app: FastAPI):
    print("子应用启动")
    # 初始化子应用资源
    yield
    print("子应用关闭")
    # 清理子应用资源

sub_app = FastAPI(lifespan=sub_lifespan, title="子应用")

@sub_app.get("/")
async def sub_root():
    return {"message": "子应用"}

# 主应用的生命周期
@asynccontextmanager
async def main_lifespan(app: FastAPI):
    print("主应用启动")
    yield
    print("主应用关闭")

app = FastAPI(lifespan=main_lifespan, title="主应用")
app.mount("/sub", sub_app)

执行顺序：

1. 主应用启动
2. 子应用启动（首次访问时或启动时）
3. ... 处理请求 ...
4. 子应用关闭
5. 主应用关闭

root_path 的作用

当挂载子应用时，FastAPI 自动处理 root_path，确保文档和路由正确工作：

from fastapi import FastAPI, Request

app = FastAPI(title="主应用")
sub_app = FastAPI(title="子应用")

@sub_app.get("/info")
async def info(request: Request):
    # request.scope["root_path"] = "/sub"
    return {
        "root_path": request.scope.get("root_path"),
        "url": str(request.url)
    }

app.mount("/sub", sub_app)

访问 /sub/info 返回：

{
    "root_path": "/sub",
    "url": "http://localhost:8000/sub/info"
}

挂载其他 ASGI 应用

除了 FastAPI 子应用，还可以挂载其他 ASGI 应用：

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import JSONResponse

# 主应用
app = FastAPI()

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

# 挂载自定义 ASGI 应用
async def custom_app(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'application/json']],
    })
    await send({
        'type': 'http.response.body',
        'body': b'{"message": "Custom ASGI app"}',
    })

app.mount("/custom", custom_app)

注意事项

1. 中间件隔离：子应用不会继承主应用的中间件，需要独立配置
2. 依赖注入隔离：子应用的依赖注入独立于主应用
3. 生命周期独立：子应用有自己的生命周期事件
4. 文档独立：每个子应用有独立的 OpenAPI 文档

小结

本章我们学习了：

1. 路由的基本定义方式
2. 路径参数和查询参数的使用
3. 参数验证和类型转换
4. 使用 APIRouter 组织路由
5. 响应模型和状态码配置
6. 子应用与挂载：创建独立的子应用，拥有独立的文档和配置

练习

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