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

FastAPI 凭借其高性能、易用性和自动生成 API 文档的特性，已成为开发 Web API 的热门选择。本文将通过一个完整的待办事项（Todo）管理项目，详细讲解如何使用 FastAPI 实现路由的增删改查（CRUD）功能。项目将涵盖从初始化到部署的全流程，并提供可运行的代码示例。

一、项目初始化与依赖安装

1.1 创建项目结构

首先，创建一个标准的 Python 项目结构：

todo_api/
├── main.py            # 主入口文件
├── models.py          # 数据模型定义
├── schemas.py         # 数据验证模型
├── crud.py            # 数据库操作逻辑
├── database.py        # 数据库连接
└── requirements.txt   # 依赖列表

1.2 安装依赖

使用 pip 安装 FastAPI 和 Uvicorn（ASGI 服务器）：

pip install fastapi uvicorn sqlalchemy

对于数据库操作，推荐使用 SQLAlchemy 作为 ORM 工具。如果需要使用其他数据库（如 PostgreSQL、MySQL），需安装对应的驱动（如 psycopg2-binary、pymysql）。

二、数据库模型与数据验证

2.1 定义数据库模型

在 models.py 中定义待办事项的数据库模型：

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
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)
    created_at = Column(DateTime, default=datetime.utcnow)

2.2 定义 Pydantic 模型

在 schemas.py 中定义用于数据验证的 Pydantic 模型：

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

TodoBase 定义了基础字段，TodoCreate 和 TodoUpdate 分别用于创建和更新操作，Todo 用于返回完整的待办事项数据。

三、数据库连接与 CRUD 操作

3.1 数据库连接

在 database.py 中配置数据库连接：

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

这里使用 SQLite 作为开发数据库，生产环境可替换为其他数据库。

3.2 实现 CRUD 逻辑

在 crud.py 中实现待办事项的增删改查操作：

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(**todo.dict())
    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 = db.query(models.Todo).filter(models.Todo.id == todo_id).first()
    if db_todo:
        for key, value in todo.dict(exclude_unset=True).items():
            setattr(db_todo, key, value)
        db.commit()
        db.refresh(db_todo)
    return db_todo
def delete_todo(db: Session, todo_id: int):
    db_todo = db.query(models.Todo).filter(models.Todo.id == todo_id).first()
    if db_todo:
        db.delete(db_todo)
        db.commit()
    return db_todo

四、实现路由与 API 端点

4.1 创建路由

在 main.py 中实现待办事项的路由：

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
@app.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)
@app.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
@app.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
@app.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
@app.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"}

4.2 路由解析

• 创建待办事项：POST /todos/
  接收 TodoCreate 模型数据，返回创建的待办事项。

• 获取所有待办事项：GET /todos/
  支持分页参数 skip 和 limit，返回待办事项列表。

• 获取单个待办事项：GET /todos/{todo_id}
  根据 ID 获取待办事项，未找到时返回 404 错误。

• 更新待办事项：PUT /todos/{todo_id}
  接收 TodoUpdate 模型数据，更新指定 ID 的待办事项。

• 删除待办事项：DELETE /todos/{todo_id}
  删除指定 ID 的待办事项，返回成功消息。

五、运行与测试 API

5.1 启动服务

使用 Uvicorn 运行 FastAPI 应用：

uvicorn main:app --reload

服务启动后，访问 http://127.0.0.1:8000/docs 可查看自动生成的交互式 API 文档。

5.2 测试 API

使用 curl 或 Postman 测试 API 端点：

创建待办事项

curl -X POST "http://127.0.0.1:8000/todos/" -H "Content-Type: application/json" -d '{"title": "Learn FastAPI", "description": "Complete the tutorial"}'

获取所有待办事项

curl "http://127.0.0.1:8000/todos/"

更新待办事项

curl -X PUT "http://127.0.0.1:8000/todos/1" -H "Content-Type: application/json" -d '{"completed": true}'

删除待办事项

curl -X DELETE "http://127.0.0.1:8000/todos/1"

六、进阶优化与部署

6.1 添加依赖注入

使用 FastAPI 的 Depends 实现依赖注入，如数据库会话管理，已在上文中体现。

6.2 添加异常处理

自定义异常处理器，统一返回错误格式：

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
    )

6.3 部署到生产环境

推荐使用 Docker 容器化部署：

FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

构建并运行容器：

docker build -t todo-api .
docker run -d -p 8000:8000 todo-api

七、总结与扩展

本文通过一个完整的待办事项管理项目，详细讲解了 FastAPI 实现 CRUD 路由的全流程。关键点包括：

1. 项目结构：模块化设计，分离模型、路由和数据库逻辑。
2. 数据验证：使用 Pydantic 模型确保数据有效性。
3. 数据库操作：通过 SQLAlchemy 实现 ORM 操作。
4. 路由实现：定义清晰的 RESTful 接口。
5. 部署优化：使用 Docker 容器化部署。

扩展方向

1. 添加用户认证：使用 JWT 或 OAuth2 实现权限控制。
2. 支持多数据库：通过配置动态切换数据库。
3. 添加缓存：使用 Redis 缓存频繁访问的数据。
4. 日志与监控：集成 Prometheus 和 Grafana 实现监控。

通过本文的指导，读者可以快速掌握 FastAPI 开发 Web API 的核心技能，并构建出可扩展的生产级应用。