FastAPI 实战：从零构建待办事项 API 的 CRUD 操作指南

FastAPI 作为现代 Python Web 框架的代表，凭借其高性能、自动生成文档和异步支持等特性，已成为开发 RESTful API 的首选工具。本文将以待办事项（Todo）管理为例，深入讲解如何使用 FastAPI 实现完整的增删改查（CRUD）路由，覆盖从项目初始化到功能实现的完整流程。

一、项目初始化与基础配置

1. 环境准备与依赖安装

开发 FastAPI 项目前，需确保 Python 版本 ≥3.7（推荐 3.9+），并通过 pip 安装核心依赖：

pip install fastapi uvicorn[standard] pydantic sqlalchemy

• fastapi：核心框架
• uvicorn：ASGI 服务器
• pydantic：数据验证与序列化
• sqlalchemy：ORM 工具（可选，用于数据库操作）

2. 项目结构规划

合理的项目结构能提升代码可维护性。推荐采用以下分层架构：

todo_api/
├── main.py            # 应用入口
├── models/            # 数据模型
│   └── todo.py        # Todo 模型定义
├── schemas/           # 数据验证 schema
│   └── todo.py        # Todo 请求/响应模型
├── routers/           # 路由模块
│   └── todos.py       # Todo 相关路由
├── database.py        # 数据库连接（可选）
└── requirements.txt   # 依赖清单

二、核心功能实现：CRUD 路由设计

1. 数据模型与 Schema 定义

数据模型（models/todo.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)

请求/响应 Schema（schemas/todo.py）

from pydantic import BaseModel
class TodoBase(BaseModel):
    title: str
    description: str = None
    completed: bool = False
class TodoCreate(TodoBase):
    pass
class TodoUpdate(TodoBase):
    pass
class Todo(TodoBase):
    id: int
    class Config:
        orm_mode = True  # 允许从 ORM 对象转换

2. 路由实现（routers/todos.py）

初始化路由与依赖注入

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from ..database import get_db  # 假设已实现数据库会话工厂
from ..models.todo import Todo as TodoModel
from ..schemas.todo import TodoCreate, TodoUpdate, Todo as TodoSchema
router = APIRouter()

创建（Create）

@router.post("/", response_model=TodoSchema)
def create_todo(
    todo: TodoCreate,
    db: Session = Depends(get_db)
):
    db_todo = TodoModel(**todo.dict())
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return db_todo

关键点：

• 使用 TodoCreate Schema 验证输入数据
• 通过依赖注入获取数据库会话
• 返回完整的 TodoSchema 对象

读取（Read）

获取所有待办事项：

@router.get("/", response_model=List[TodoSchema])
def read_todos(
    db: Session = Depends(get_db),
    skip: int = 0,
    limit: int = 100
):
    return db.query(TodoModel).offset(skip).limit(limit).all()

获取单个待办事项：

@router.get("/{todo_id}", response_model=TodoSchema)
def read_todo(
    todo_id: int,
    db: Session = Depends(get_db)
):
    db_todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return db_todo

更新（Update）

@router.put("/{todo_id}", response_model=TodoSchema)
def update_todo(
    todo_id: int,
    todo_update: TodoUpdate,
    db: Session = Depends(get_db)
):
    db_todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    update_data = todo_update.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_todo, key, value)
    db.commit()
    db.refresh(db_todo)
    return db_todo

优化点：

• 使用 exclude_unset=True 忽略未提供的字段
• 通过 setattr 动态更新属性

删除（Delete）

@router.delete("/{todo_id}")
def delete_todo(
    todo_id: int,
    db: Session = Depends(get_db)
):
    db_todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    db.delete(db_todo)
    db.commit()
    return {"message": "Todo deleted successfully"}

3. 主应用集成（main.py）

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

三、进阶优化与最佳实践

1. 异步支持

FastAPI 原生支持异步路由，可提升 I/O 密集型操作性能：

from sqlalchemy.ext.asyncio import AsyncSession
from .database import get_async_db  # 异步数据库会话
@router.post("/async/", response_model=TodoSchema)
async def create_todo_async(
    todo: TodoCreate,
    db: AsyncSession = Depends(get_async_db)
):
    db_todo = TodoModel(**todo.dict())
    db.add(db_todo)
    await db.commit()
    await db.refresh(db_todo)
    return db_todo

2. 分页与排序

增强读取接口的灵活性：

from fastapi import Query
@router.get("/", response_model=List[TodoSchema])
def read_todos(
    db: Session = Depends(get_db),
    skip: int = Query(0, ge=0),
    limit: int = Query(100, le=500),
    sort_by: str = Query("id", description="Field to sort by"),
    sort_order: str = Query("asc", description="Sort order (asc/desc)")
):
    query = db.query(TodoModel)
    # 动态排序
    if sort_order == "desc":
        query = query.order_by(getattr(TodoModel, sort_by).desc())
    else:
        query = query.order_by(getattr(TodoModel, sort_by).asc())
    return query.offset(skip).limit(limit).all()

3. 错误处理与日志

自定义异常处理器：

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},
    )

四、测试与部署

1. 单元测试示例

使用 pytest 和 httpx 测试 API：

import pytest
from httpx import AsyncClient
from main import app
@pytest.mark.anyio
async def test_create_todo():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/todos/", json={
            "title": "Test Todo",
            "description": "Test Description"
        })
    assert response.status_code == 200
    assert response.json()["title"] == "Test Todo"

2. 生产部署建议

• ASGI 服务器：推荐 uvicorn 或 gunicorn + uvicorn 工作模式
• 进程管理：使用 systemd 或 Docker 容器化部署
• 反向代理：Nginx 配置示例：

location / {
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
}

五、总结与扩展

本文通过待办事项管理案例，系统展示了 FastAPI 实现 CRUD 路由的核心方法。关键收获包括：

1. 分层架构：分离路由、模型和 Schema，提升可维护性
2. 依赖注入：通过 Depends 管理数据库会话等共享资源
3. 数据验证：利用 Pydantic Schema 确保输入输出合法性
4. 异步支持：优化高并发场景下的性能表现

扩展方向：

• 添加用户认证（JWT/OAuth2）
• 实现事件驱动（如 WebSocket 通知）
• 集成缓存（Redis）
• 添加 API 版本控制

FastAPI 的简洁设计与强大功能，使其成为构建现代 Web API 的理想选择。通过本文的实践，开发者可快速掌握从路由设计到生产部署的全流程，为后续复杂项目开发奠定坚实基础。