FastAPI 实战：待办事项路由的增删改查全流程指南

FastAPI 作为一款现代化的 Python Web 框架，以其高性能、自动生成 API 文档和类型提示支持等特性，成为开发 Web API 的理想选择。本文将通过一个完整的待办事项（Todo）项目，详细阐述如何使用 FastAPI 实现路由的增删改查（CRUD）功能，帮助开发者快速掌握 FastAPI 的核心开发技巧。

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

1.1 环境搭建

首先，我们需要创建一个 Python 虚拟环境，并安装 FastAPI 及其依赖。推荐使用 pipenv 或 poetry 进行依赖管理，这里以 pipenv 为例：

# 创建虚拟环境并安装 FastAPI
pipenv install fastapi uvicorn[standard]
# 安装数据库依赖（以 SQLite 为例）
pipenv install sqlalchemy databases[sqlite]

1.2 项目结构规划

合理的项目结构有助于代码的可维护性。一个典型的 FastAPI 项目结构如下：

todo_app/
├── main.py          # 应用入口
├── models.py        # 数据模型定义
├── schemas.py       # 数据验证模型
├── crud.py          # 数据库操作逻辑
├── database.py      # 数据库连接与会话管理
└── routers/         # 路由模块
    └── todos.py     # 待办事项路由

二、数据模型与数据库集成

2.1 定义数据模型

在 models.py 中，我们使用 SQLAlchemy 定义待办事项的数据模型：

from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Todo(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String)
    completed = Column(Boolean, default=False)

2.2 数据库连接与会话管理

在 database.py 中，我们配置数据库连接并创建会话工厂：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from .models import Base
DATABASE_URL = "sqlite:///./todo.db"
engine = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)

2.3 创建数据库操作函数

在 crud.py 中，我们封装待办事项的 CRUD 操作：

from sqlalchemy.orm import Session
from . import models, schemas
def get_todo(db: Session, todo_id: int):
    return db.query(models.Todo).filter(models.Todo.id == todo_id).first()
def get_todos(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Todo).offset(skip).limit(limit).all()
def create_todo(db: Session, todo: schemas.TodoCreate):
    db_todo = models.Todo(title=todo.title, description=todo.description)
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return db_todo
def update_todo(db: Session, todo_id: int, todo: schemas.TodoUpdate):
    db_todo = get_todo(db, todo_id)
    if db_todo:
        db_todo.title = todo.title or db_todo.title
        db_todo.description = todo.description or db_todo.description
        db_todo.completed = todo.completed if todo.completed is not None else db_todo.completed
        db.commit()
    return db_todo
def delete_todo(db: Session, todo_id: int):
    db_todo = get_todo(db, todo_id)
    if db_todo:
        db.delete(db_todo)
        db.commit()
    return db_todo

三、路由设计与实现

3.1 定义请求与响应模型

在 schemas.py 中，我们使用 Pydantic 定义请求和响应的数据模型：

from pydantic import BaseModel
class TodoBase(BaseModel):
    title: str
    description: str | None = None
class TodoCreate(TodoBase):
    pass
class TodoUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    completed: bool | None = None
class Todo(TodoBase):
    id: int
    completed: bool
    class Config:
        orm_mode = True

3.2 实现待办事项路由

在 routers/todos.py 中，我们定义待办事项的 CRUD 路由：

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from .. import crud, schemas
from ..database import SessionLocal
router = APIRouter()
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
@router.get("/todos/", response_model=list[schemas.Todo])
def read_todos(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    todos = crud.get_todos(db, skip=skip, limit=limit)
    return todos
@router.post("/todos/", response_model=schemas.Todo)
def create_todo(todo: schemas.TodoCreate, db: Session = Depends(get_db)):
    return crud.create_todo(db=db, todo=todo)
@router.get("/todos/{todo_id}", response_model=schemas.Todo)
def read_todo(todo_id: int, db: Session = Depends(get_db)):
    db_todo = crud.get_todo(db, todo_id=todo_id)
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return db_todo
@router.put("/todos/{todo_id}", response_model=schemas.Todo)
def update_todo(todo_id: int, todo: schemas.TodoUpdate, db: Session = Depends(get_db)):
    db_todo = crud.update_todo(db, todo_id=todo_id, todo=todo)
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return db_todo
@router.delete("/todos/{todo_id}")
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    db_todo = crud.delete_todo(db, todo_id=todo_id)
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return {"message": "Todo deleted successfully"}

3.3 集成路由到主应用

在 main.py 中，我们集成路由并启动应用：

from fastapi import FastAPI
from .routers import todos
app = FastAPI()
app.include_router(todos.router)
# 启动命令：uvicorn main:app --reload

四、测试与验证

4.1 使用 HTTP 客户端测试

FastAPI 自动生成交互式 API 文档（访问 /docs），我们可以使用它进行测试：

1. 创建待办事项：

   • 方法：POST
   • 路径：/todos/
   • 请求体：

     {
       "title": "学习 FastAPI",
       "description": "完成待办事项路由的增删改查"
     }

2. 查询待办事项：

   • 方法：GET
   • 路径：/todos/

3. 更新待办事项：

   • 方法：PUT
   • 路径：/todos/{todo_id}
   • 请求体：

     {
       "completed": true
     }

4. 删除待办事项：

   • 方法：DELETE
   • 路径：/todos/{todo_id}

4.2 编写单元测试

使用 pytest 编写单元测试，验证路由逻辑的正确性：

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app
from database import Base
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)
def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
def test_create_todo():
    response = client.post(
        "/todos/",
        json={"title": "测试待办", "description": "测试描述"},
    )
    assert response.status_code == 200
    assert response.json()["title"] == "测试待办"
def test_read_todos():
    response = client.get("/todos/")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

五、优化与扩展建议

5.1 性能优化

1. 异步支持：FastAPI 原生支持异步，可以将数据库操作改为异步（如使用 asyncpg 和 sqlalchemy[asyncio]）。
2. 缓存层：引入 Redis 缓存频繁查询的数据，减少数据库压力。
3. 分页优化：在查询大量数据时，实现更高效的分页逻辑。

5.2 功能扩展

1. 用户认证：集成 JWT 或 OAuth2 实现用户认证。
2. 标签系统：为待办事项添加标签，支持按标签筛选。
3. 定时任务：使用 Celery 实现待办事项的定时提醒。

5.3 部署建议

1. 容器化：使用 Docker 打包应用，便于部署。
2. CI/CD：集成 GitHub Actions 或 GitLab CI 实现自动化测试和部署。
3. 监控：使用 Prometheus 和 Grafana 监控应用性能。

六、总结

通过本文，我们详细讲解了如何使用 FastAPI 快速开发一个支持增删改查的待办事项 Web API 项目。从环境配置、数据模型定义、路由设计到测试验证，我们覆盖了 FastAPI 开发的核心流程。FastAPI 的高性能和易用性使其成为开发 Web API 的理想选择，而合理的项目结构和代码组织则能显著提升开发效率和代码质量。希望本文能为开发者提供实用的指导，助力快速掌握 FastAPI 的开发技巧。