FastAPI 实战：构建待办事项 Web API 的增删改查全流程

2025年11月14日 互联网

一、FastAPI 简介与项目初始化

FastAPI 是一个基于 Python 的现代 Web 框架，用于快速构建 API。它利用 Python 类型注解实现自动文档生成（Swagger UI 和 ReDoc），并支持异步请求处理，性能接近 Node.js 和 Go。相比 Flask 和 Django，FastAPI 更轻量且适合 API 开发。

1.1 环境准备

首先，确保已安装 Python 3.7+ 和 pip。创建虚拟环境并安装 FastAPI 和 Uvicorn（ASGI 服务器）：

  1. python -m venv venv
  2. source venv/bin/activate  # Linux/macOS
  3. venv\Scripts\activate     # Windows
  4. pip install fastapi uvicorn

1.2 项目结构

推荐结构如下：

  1. todo_api/
  2. ├── main.py            # 入口文件
  3. ├── models.py          # 数据模型
  4. ├── schemas.py         # 请求/响应模型（Pydantic）
  5. ├── crud.py            # 数据库操作
  6. └── requirements.txt   # 依赖文件

二、定义数据模型与 Pydantic 模式

2.1 数据模型（models.py）

使用 SQLModel（结合 SQLAlchemy 和 Pydantic）定义待办事项模型：

  1. from sqlmodel import SQLModel, Field
  2. from typing import Optional
  4. class Todo(SQLModel, table=True):
  5.     id: Optional[int] = Field(default=None, primary_key=True)
  6.     title: str = Field(max_length=100)
  7.     description: Optional[str] = Field(default=None)
  8.     completed: bool = Field(default=False)

2.2 请求/响应模式（schemas.py）

使用 Pydantic 定义数据验证模式：

  1. from pydantic import BaseModel
  2. from typing import Optional
  4. class TodoBase(BaseModel):
  5.     title: str
  6.     description: Optional[str] = None
  7.     completed: bool = False
  9. class TodoCreate(TodoBase):
  10.     pass
  12. class TodoUpdate(TodoBase):
  13.     pass
  15. class Todo(TodoBase):
  16.     id: int
  18.     class Config:
  19.         orm_mode = True  # 支持 SQLModel 对象转换

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

3.1 数据库连接与初始化

使用 sqlmodel.Session 管理数据库会话：

  1. from sqlmodel import SQLModel, create_engine, Session
  2. from typing import Optional
  3. from .models import Todo
  5. # SQLite 内存数据库（生产环境替换为 PostgreSQL/MySQL）
  6. engine = create_engine("sqlite:///todos.db")
  8. def init_db():
  9.     SQLModel.metadata.create_all(engine)

3.2 核心 CRUD 函数

  1. def get_todos(session: Session):
  2.     return session.query(Todo).all()
  4. def get_todo(session: Session, todo_id: int):
  5.     return session.query(Todo).filter(Todo.id == todo_id).first()
  7. def create_todo(session: Session, todo: TodoCreate):
  8.     db_todo = Todo.from_orm(todo)
  9.     session.add(db_todo)
  10.     session.commit()
  11.     session.refresh(db_todo)
  12.     return db_todo
  14. def update_todo(session: Session, todo_id: int, todo: TodoUpdate):
  15.     db_todo = session.query(Todo).filter(Todo.id == todo_id).first()
  16.     if db_todo:
  17.         todo_data = todo.dict(exclude_unset=True)
  18.         for key, value in todo_data.items():
  19.             setattr(db_todo, key, value)
  20.         session.commit()
  21.     return db_todo
  23. def delete_todo(session: Session, todo_id: int):
  24.     db_todo = session.query(Todo).filter(Todo.id == todo_id).first()
  25.     if db_todo:
  26.         session.delete(db_todo)
  27.         session.commit()
  28.     return db_todo is not None

四、构建 FastAPI 路由

4.1 初始化 FastAPI 应用

在 main.py 中：

  1. from fastapi import FastAPI, Depends, HTTPException
  2. from sqlmodel import Session
  3. from .crud import init_db, get_todos, get_todo, create_todo, update_todo, delete_todo
  4. from .models import Todo
  5. from .schemas import TodoCreate, TodoUpdate
  6. from .database import SessionLocal  # 需在 database.py 中定义
  8. app = FastAPI()
  10. # 依赖项：获取数据库会话
  11. def get_db():
  12.     with SessionLocal() as session:
  13.         yield session
  15. @app.on_event("startup")
  16. async def startup():
  17.     init_db()

4.2 实现增删改查路由

  1. # 创建待办事项
  2. @app.post("/todos/", response_model=Todo)
  3. def create_todo_route(todo: TodoCreate, db: Session = Depends(get_db)):
  4.     return create_todo(db, todo)
  6. # 获取所有待办事项
  7. @app.get("/todos/", response_model=list[Todo])
  8. def read_todos(db: Session = Depends(get_db)):
  9.     return get_todos(db)
  11. # 获取单个待办事项
  12. @app.get("/todos/{todo_id}", response_model=Todo)
  13. def read_todo(todo_id: int, db: Session = Depends(get_db)):
  14.     db_todo = get_todo(db, todo_id)
  15.     if db_todo is None:
  16.         raise HTTPException(status_code=404, detail="Todo not found")
  17.     return db_todo
  19. # 更新待办事项
  20. @app.put("/todos/{todo_id}", response_model=Todo)
  21. def update_todo_route(todo_id: int, todo: TodoUpdate, db: Session = Depends(get_db)):
  22.     db_todo = update_todo(db, todo_id, todo)
  23.     if db_todo is None:
  24.         raise HTTPException(status_code=404, detail="Todo not found")
  25.     return db_todo
  27. # 删除待办事项
  28. @app.delete("/todos/{todo_id}")
  29. def delete_todo_route(todo_id: int, db: Session = Depends(get_db)):
  30.     success = delete_todo(db, todo_id)
  31.     if not success:
  32.         raise HTTPException(status_code=404, detail="Todo not found")
  33.     return {"message": "Todo deleted successfully"}

五、运行与测试 API

5.1 启动服务

  1. uvicorn main:app --reload

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

5.2 测试 API

使用 curl 或 HTTPie 测试：

  • 创建待办事项：

    1. curl -X POST "http://127.0.0.1:8000/todos/" -H "Content-Type: application/json" -d '{"title": "Learn FastAPI", "description": "Build a TODO API"}'

  • 获取所有待办事项：

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

  • 更新待办事项：

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

  • 删除待办事项：

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

5.3 自动生成的文档

访问 http://127.0.0.1:8000/docs（Swagger UI）或 http://127.0.0.1:8000/redoc（ReDoc）查看交互式文档。

六、进阶优化建议

  1. 数据库持久化：替换 SQLite 为 PostgreSQL/MySQL，并配置连接池。
  2. 异步支持：使用 SQLModel 的异步版本（需 asyncpg）。
  3. 认证与授权：集成 JWT 或 OAuth2。
  4. 分页与过滤：在 get_todos 中添加分页参数（如 skip、limit）。
  5. 单元测试：使用 pytest 编写测试用例。

七、总结

通过本文，您已掌握如何使用 FastAPI 快速开发一个完整的待办事项 Web API，包括：

  • 项目初始化与环境配置。
  • 数据模型与 Pydantic 模式定义。
  • 核心 CRUD 操作实现。
  • 路由设计与依赖注入。
  • 服务运行与 API 测试。

FastAPI 的高性能和自动文档特性使其成为构建现代 Web API 的理想选择。

最新文章
热门标签