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

一、FastAPI 快速开发 Web API 的核心优势

FastAPI 作为现代 Python Web 框架，凭借其高性能（基于 Starlette 和 Pydantic）、自动生成 API 文档（Swagger UI 和 ReDoc）和类型提示支持，成为开发 RESTful API 的首选工具。其核心优势在于：

1. 开发效率：通过类型注解自动生成验证逻辑，减少重复代码。
2. 异步支持：原生支持 async/await，适合高并发场景。
3. 标准兼容：基于 OpenAPI 和 JSON Schema，接口规范清晰。

以待办事项（Todo）管理为例，FastAPI 可快速实现增删改查（CRUD）功能，覆盖从路由设计到数据库交互的全流程。

二、项目初始化与依赖配置

1. 环境准备

使用 pip 安装核心依赖：

pip install fastapi uvicorn[standard] sqlalchemy  # 基础依赖
pip install python-dotenv  # 环境变量管理（可选）

2. 项目结构

推荐分层架构：

todo_api/
├── main.py          # 入口文件
├── models.py        # 数据模型
├── schemas.py       # 请求/响应模型
├── crud.py          # 数据库操作
├── database.py      # 数据库连接
└── routers/         # 路由模块
    └── todos.py     # 待办事项路由

3. 数据库连接（以 SQLite 为例）

在 database.py 中配置异步数据库引擎：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite+aiosqlite:///./todo.db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

三、待办事项模型定义

1. SQLAlchemy 模型（models.py）

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. Pydantic 模型（schemas.py）

from pydantic import BaseModel
class TodoBase(BaseModel):
    title: str
    description: str | None = None
    completed: bool = False
class TodoCreate(TodoBase):
    pass
class Todo(TodoBase):
    id: int
    class Config:
        orm_mode = True  # 支持 SQLAlchemy 模型转换

四、CRUD 操作实现（crud.py）

1. 基础操作函数

from sqlalchemy.ext.asyncio import AsyncSession
from .models import Todo
from .schemas import TodoCreate
async def create_todo(db: AsyncSession, todo: TodoCreate):
    db_todo = Todo(**todo.dict())
    db.add(db_todo)
    await db.commit()
    await db.refresh(db_todo)
    return db_todo
async def get_todo(db: AsyncSession, todo_id: int):
    return await db.get(Todo, todo_id)
async def get_todos(db: AsyncSession, skip: int = 0, limit: int = 100):
    return await db.scalars(
        db.query(Todo).offset(skip).limit(limit).all()
    )
async def update_todo(db: AsyncSession, todo_id: int, todo: TodoCreate):
    db_todo = await db.get(Todo, todo_id)
    if db_todo:
        for key, value in todo.dict().items():
            setattr(db_todo, key, value)
        await db.commit()
    return db_todo
async def delete_todo(db: AsyncSession, todo_id: int):
    db_todo = await db.get(Todo, todo_id)
    if db_todo:
        await db.delete(db_todo)
        await db.commit()
    return db_todo

五、路由设计与实现（routers/todos.py）

1. 依赖注入与路由组织

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from ..database import AsyncSessionLocal
from ..crud import create_todo, get_todo, get_todos, update_todo, delete_todo
from ..schemas import Todo, TodoCreate
router = APIRouter()
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session
@router.post("/", response_model=Todo)
async def create_todo_item(todo: TodoCreate, db: AsyncSession = Depends(get_db)):
    return await create_todo(db, todo)
@router.get("/{todo_id}", response_model=Todo)
async def read_todo(todo_id: int, db: AsyncSession = Depends(get_db)):
    db_todo = await get_todo(db, todo_id)
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return db_todo
@router.get("/", response_model=list[Todo])
async def read_todos(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)):
    todos = await get_todos(db, skip=skip, limit=limit)
    return todos
@router.put("/{todo_id}", response_model=Todo)
async def update_todo_item(todo_id: int, todo: TodoCreate, db: AsyncSession = Depends(get_db)):
    updated_todo = await update_todo(db, todo_id, todo)
    if updated_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return updated_todo
@router.delete("/{todo_id}")
async def delete_todo_item(todo_id: int, db: AsyncSession = Depends(get_db)):
    deleted_todo = await delete_todo(db, todo_id)
    if deleted_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return {"message": "Todo deleted successfully"}

六、主程序集成（main.py）

from fastapi import FastAPI
from .routers import todos
app = FastAPI()
app.include_router(todos.router, prefix="/todos", tags=["todos"])
@app.get("/")
async def read_root():
    return {"message": "Welcome to the Todo API"}

七、运行与测试

1. 启动服务

uvicorn main:app --reload

2. 测试接口

使用 curl 或 Postman 测试：

• 创建待办事项：

  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 '{"title":"Learn FastAPI","description":"Master CRUD operations","completed":true}'

• 删除待办事项：

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

八、进阶优化建议

1. 分页查询：在 get_todos 中实现 skip 和 limit 参数。
2. 错误处理：自定义异常处理器统一返回格式。
3. 认证授权：集成 OAuth2 或 APIKey 保护接口。
4. 日志记录：使用 logging 模块记录请求和操作。
5. 部署优化：配置 Gunicorn + Uvicorn 用于生产环境。

九、总结

通过 FastAPI 实现待办事项的增删改查功能，开发者可以快速掌握以下核心技能：

1. 路由设计：基于 RESTful 原则组织 API 端点。
2. 依赖注入：通过 Depends 管理数据库会话。
3. 异步编程：使用 async/await 处理 I/O 密集型操作。
4. 数据验证：利用 Pydantic 模型自动校验请求体。

此项目可作为学习 FastAPI 的入门范例，也可扩展为完整的任务管理系统。实际开发中，可进一步集成数据库迁移工具（如 Alembic）、消息队列（如 Celery）或前端框架（如 React/Vue）构建全栈应用。