FastAPI 实战：待办事项管理系统的 CRUD 路由设计与实现

FastAPI 作为基于 Python 的高性能 Web 框架，凭借其类型提示支持、自动生成 API 文档和异步处理能力，已成为开发 RESTful API 的首选工具之一。本文将以待办事项（Todo）管理系统为例，详细演示如何使用 FastAPI 实现完整的 CRUD（增删改查）路由，覆盖从模型定义到路由设计的全流程，帮助开发者快速掌握 FastAPI 的核心开发模式。

一、项目初始化与环境配置

1.1 创建项目结构

使用 fastapi 和 uvicorn 构建项目时，建议采用模块化设计。典型的项目结构如下：

todo_api/
├── main.py          # 应用入口
├── models.py        # 数据模型定义
├── schemas.py       # 请求/响应模型（Pydantic）
├── routers/         # 路由模块
│   └── todos.py     # 待办事项路由
└── requirements.txt # 依赖列表

1.2 安装依赖

通过 pip 安装必要依赖：

pip install fastapi uvicorn pydantic sqlmodel

• fastapi：核心框架
• uvicorn：ASGI 服务器
• pydantic：数据验证与序列化
• sqlmodel（可选）：结合 SQLAlchemy 与 Pydantic 的 ORM 工具

二、数据模型与请求模型设计

2.1 定义待办事项模型

使用 SQLModel（或直接使用 Pydantic）定义数据模型：

# models.py
from sqlmodel import SQLModel, Field
from typing import Optional
class Todo(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(max_length=100)
    description: Optional[str] = Field(default=None)
    completed: bool = Field(default=False)

2.2 定义请求/响应模型

通过 Pydantic 模型规范 API 输入输出：

# schemas.py
from pydantic import BaseModel
from typing import Optional
class TodoCreate(BaseModel):
    title: str
    description: Optional[str] = None
class TodoUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    completed: Optional[bool] = None
class TodoResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    completed: bool

三、路由设计与 CRUD 实现

3.1 创建路由模块

在 routers/todos.py 中实现路由逻辑：

# routers/todos.py
from fastapi import APIRouter, HTTPException
from typing import List
from ..models import Todo
from ..schemas import TodoCreate, TodoUpdate, TodoResponse
# 模拟数据库
fake_db: List[Todo] = []
router = APIRouter(prefix="/todos", tags=["todos"])
@router.post("/", response_model=TodoResponse)
def create_todo(todo: TodoCreate):
    """创建待办事项"""
    new_id = len(fake_db) + 1
    todo_item = Todo(id=new_id, **todo.dict())
    fake_db.append(todo_item)
    return todo_item
@router.get("/", response_model=List[TodoResponse])
def read_todos():
    """获取所有待办事项"""
    return fake_db
@router.get("/{todo_id}", response_model=TodoResponse)
def read_todo(todo_id: int):
    """获取单个待办事项"""
    for todo in fake_db:
        if todo.id == todo_id:
            return todo
    raise HTTPException(status_code=404, detail="Todo not found")
@router.put("/{todo_id}", response_model=TodoResponse)
def update_todo(todo_id: int, todo_update: TodoUpdate):
    """更新待办事项"""
    for todo in fake_db:
        if todo.id == todo_id:
            update_data = todo_update.dict(exclude_unset=True)
            for field, value in update_data.items():
                setattr(todo, field, value)
            return todo
    raise HTTPException(status_code=404, detail="Todo not found")
@router.delete("/{todo_id}")
def delete_todo(todo_id: int):
    """删除待办事项"""
    for i, todo in enumerate(fake_db):
        if todo.id == todo_id:
            del fake_db[i]
            return {"message": "Todo deleted successfully"}
    raise HTTPException(status_code=404, detail="Todo not found")

3.2 路由集成与主应用

在 main.py 中注册路由：

# main.py
from fastapi import FastAPI
from routers.todos import router as todos_router
app = FastAPI()
app.include_router(todos_router)
@app.get("/")
def read_root():
    return {"message": "Welcome to Todo API"}

四、运行与测试 API

4.1 启动服务

使用 uvicorn 运行应用：

uvicorn main:app --reload

服务默认运行在 http://127.0.0.1:8000。

4.2 测试 API

FastAPI 自动生成交互式文档（Swagger UI），访问 http://127.0.0.1:8000/docs 即可测试所有路由：

• 创建待办事项：POST /todos/
• 获取所有待办事项：GET /todos/
• 获取单个待办事项：GET /todos/{todo_id}
• 更新待办事项：PUT /todos/{todo_id}
• 删除待办事项：DELETE /todos/{todo_id}

五、进阶优化建议

5.1 数据库集成

将内存数据库替换为真实数据库（如 SQLite、PostgreSQL）：

# 初始化 SQLModel 数据库
from sqlmodel import Session, SQLModel, create_engine
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

5.2 添加身份验证

集成 JWT 或 OAuth2 实现 API 安全：

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_token(token: str = Depends(oauth2_scheme)):
    # 实际项目中需验证 token 有效性
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return token

5.3 异步支持

使用 async/await 提升性能：

@router.get("/async/{todo_id}", response_model=TodoResponse)
async def read_todo_async(todo_id: int):
    # 模拟异步数据库查询
    await asyncio.sleep(0.1)
    for todo in fake_db:
        if todo.id == todo_id:
            return todo
    raise HTTPException(status_code=404, detail="Todo not found")

六、总结与最佳实践

1. 模块化设计：将路由、模型、依赖分离，提升代码可维护性。
2. 输入验证：始终使用 Pydantic 模型验证请求数据。
3. 错误处理：通过 HTTPException 返回标准化错误响应。
4. 文档生成：利用 FastAPI 自动生成的 Swagger UI 和 ReDoc。
5. 性能优化：对 I/O 密集型操作使用异步。

通过以上步骤，开发者可以快速构建一个功能完整的待办事项管理 API，并基于 FastAPI 的特性扩展更复杂的功能。