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

2025年10月17日 互联网

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

FastAPI作为新一代Python Web框架，以其高性能、自动生成文档和类型提示支持等特性，成为构建现代Web API的首选工具。本文将通过一个完整的待办事项管理项目，深入探讨如何利用FastAPI快速实现路由的增删改查（CRUD）功能，覆盖从项目初始化到API测试的全流程。

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

1.1 创建项目结构

使用pip install fastapi uvicorn安装核心依赖后，建议采用如下项目结构：

  1. todo_api/
  2. ├── main.py          # 主入口
  3. ├── models.py        # 数据模型
  4. ├── schemas.py       # 请求/响应模型
  5. ├── crud.py          # 数据操作层
  6. └── requirements.txt # 依赖清单

这种分层设计遵循MVC模式，使代码更易于维护和扩展。

1.2 基础应用配置

在main.py中初始化FastAPI应用：

  1. from fastapi import FastAPI
  2. from fastapi.middleware.cors import CORSMiddleware
  4. app = FastAPI(title="待办事项API", version="1.0.0")
  6. # 跨域配置（开发环境必备）
  7. app.add_middleware(
  8.     CORSMiddleware,
  9.     allow_origins=["*"],
  10.     allow_methods=["*"],
  11.     allow_headers=["*"],
  12. )

二、数据模型设计

2.1 数据库模型（models.py）

使用Pydantic进行数据验证，结合SQLAlchemy（可选）进行持久化：

  1. from pydantic import BaseModel, Field
  2. from typing import Optional
  3. from datetime import datetime
  5. class TodoItem(BaseModel):
  6.     id: Optional[int] = Field(default=None, description="自动生成ID")
  7.     title: str = Field(..., min_length=3, max_length=50)
  8.     description: Optional[str] = Field(default="", max_length=200)
  9.     completed: bool = False
  10.     created_at: Optional[datetime] = Field(default_factory=datetime.now)
  11.     updated_at: Optional[datetime] = Field(default_factory=datetime.now)
  13.     class Config:
  14.         orm_mode = True  # 支持SQLAlchemy模型转换

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

分离创建和更新模型，提高接口灵活性：

  1. from pydantic import BaseModel
  3. class TodoCreate(BaseModel):
  4.     title: str
  5.     description: str = ""
  7. class TodoUpdate(BaseModel):
  8.     title: Optional[str] = None
  9.     description: Optional[str] = None
  10.     completed: Optional[bool] = None

三、CRUD路由实现

3.1 内存存储实现（开发阶段）

使用Python字典模拟数据库：

  1. from typing import Dict, Optional
  2. from fastapi import HTTPException
  3. from models import TodoItem
  5. db: Dict[int, TodoItem] = {}
  6. current_id = 1
  8. def create_todo(item: TodoItem) -> TodoItem:
  9.     global current_id
  10.     item.id = current_id
  11.     db[current_id] = item
  12.     current_id += 1
  13.     return item
  15. def get_todo(id: int) -> Optional[TodoItem]:
  16.     return db.get(id)
  18. def update_todo(id: int, updates: TodoUpdate) -> Optional[TodoItem]:
  19.     if id not in db:
  20.         raise HTTPException(status_code=404, detail="Item not found")
  22.     existing = db[id]
  23.     update_data = updates.dict(exclude_unset=True)
  24.     updated_item = existing.copy(update=update_data)
  25.     db[id] = updated_item
  26.     return updated_item
  28. def delete_todo(id: int) -> bool:
  29.     if id in db:
  30.         del db[id]
  31.         return True
  32.     return False

3.2 路由定义（main.py扩展）

  1. from fastapi import APIRouter, Path, HTTPException
  2. from schemas import TodoCreate, TodoUpdate
  3. from models import TodoItem
  4. from crud import create_todo, get_todo, update_todo, delete_todo
  6. router = APIRouter(prefix="/todos", tags=["todos"])
  8. @router.post("/", response_model=TodoItem)
  9. def create_todo_item(item: TodoCreate):
  10.     """创建新的待办事项"""
  11.     # 实际项目需在此转换模型
  12.     todo_item = TodoItem(title=item.title, description=item.description)
  13.     return create_todo(todo_item)
  15. @router.get("/{id}", response_model=TodoItem)
  16. def read_todo_item(id: int = Path(..., ge=1)):
  17.     """获取单个待办事项"""
  18.     todo = get_todo(id)
  19.     if not todo:
  20.         raise HTTPException(status_code=404, detail="Item not found")
  21.     return todo
  23. @router.put("/{id}", response_model=TodoItem)
  24. def update_todo_item(id: int, updates: TodoUpdate):
  25.     """更新待办事项"""
  26.     return update_todo(id, updates)
  28. @router.delete("/{id}")
  29. def delete_todo_item(id: int):
  30.     """删除待办事项"""
  31.     if not delete_todo(id):
  32.         raise HTTPException(status_code=404, detail="Item not found")
  33.     return {"message": "Item deleted successfully"}
  35. app.include_router(router)

四、高级功能实现

4.1 分页与过滤

  1. from fastapi import Query
  2. from typing import List
  4. @router.get("/", response_model=List[TodoItem])
  5. def read_todos(
  6.     skip: int = Query(0, ge=0),
  7.     limit: int = Query(10, le=100),
  8.     completed: Optional[bool] = None
  9. ):
  10.     """分页获取待办事项列表，支持完成状态过滤"""
  11.     items = list(db.values())
  12.     if completed is not None:
  13.         items = [item for item in items if item.completed == completed]
  14.     return items[skip : skip + limit]

4.2 依赖注入与认证

  1. from fastapi import Depends, HTTPException
  2. from fastapi.security import APIKeyHeader
  4. API_KEY = "secret-key"
  5. api_key_header = APIKeyHeader(name="X-API-Key")
  7. def get_api_key(api_key: str = Depends(api_key_header)):
  8.     if api_key != API_KEY:
  9.         raise HTTPException(status_code=403, detail="Invalid API Key")
  10.     return api_key
  12. @router.get("/secure/", response_model=List[TodoItem])
  13. def read_secure_todos(api_key: str = Depends(get_api_key)):
  14.     """需要API密钥的受保护端点"""
  15.     return list(db.values())

五、测试与文档

5.1 自动生成API文档

FastAPI自动生成Swagger UI和ReDoc文档：

  • 访问/docs查看交互式Swagger界面
  • 访问/redoc查看标准化文档

5.2 单元测试示例

  1. from fastapi.testclient import TestClient
  2. from main import app
  4. client = TestClient(app)
  6. def test_create_todo():
  7.     response = client.post(
  8.         "/todos/",
  9.         json={"title": "Test Item", "description": "Test Description"}
  10.     )
  11.     assert response.status_code == 200
  12.     assert "id" in response.json()
  14. def test_read_todo():
  15.     # 先创建测试项
  16.     create_resp = client.post(
  17.         "/todos/",
  18.         json={"title": "Read Test", "description": "Test Read"}
  19.     )
  20.     todo_id = create_resp.json()["id"]
  22.     # 读取测试
  23.     response = client.get(f"/todos/{todo_id}")
  24.     assert response.status_code == 200
  25.     assert response.json()["title"] == "Read Test"

六、生产环境部署建议

  1. 持久化存储：替换内存存储为SQLite/PostgreSQL
  2. 异步支持：使用databases库实现异步CRUD
  3. 性能优化：
    • 启用Uvicorn的--workers参数
    • 配置适当的超时设置
  4. 监控：集成Prometheus和Grafana
  5. 日志：使用结构化日志（如loguru）

七、常见问题解决方案

7.1 跨域问题

确保正确配置CORS中间件，生产环境应限制具体域名：

  1. app.add_middleware(
  2.     CORSMiddleware,
  3.     allow_origins=["https://yourdomain.com"],
  4.     allow_methods=["GET", "POST", "PUT", "DELETE"],
  5.     allow_headers=["*"],
  6. )

7.2 数据验证错误处理

FastAPI自动返回422错误，可通过异常处理器自定义：

  1. from fastapi import Request
  2. from fastapi.responses import JSONResponse
  3. from fastapi.exceptions import RequestValidationError
  5. @app.exception_handler(RequestValidationError)
  6. async def validation_exception_handler(request: Request, exc: RequestValidationError):
  7.     return JSONResponse(
  8.         status_code=422,
  9.         content={"detail": exc.errors(), "body": exc.body},
  10.     )

八、扩展建议

  1. 添加用户系统：集成OAuth2或JWT认证
  2. 附件支持：实现文件上传功能
  3. 定时任务：使用Celery处理逾期提醒
  4. WebSocket：实现实时状态更新
  5. GraphQL：通过Strawberry添加GraphQL支持

通过本文的完整实现，开发者可以快速掌握FastAPI的核心功能，构建出符合生产标准的Web API。实际项目中，建议结合具体需求进行模块化扩展，并始终遵循RESTful设计原则。FastAPI的自动文档和类型系统将显著提升开发效率和代码质量，是构建现代微服务的理想选择。

最新文章