FastAPI 待办事项API实战：从零构建CRUD路由

一、FastAPI框架优势与待办事项API场景

FastAPI作为基于Python的现代Web框架，凭借其自动生成OpenAPI文档、类型注解支持、异步请求处理等特性，成为开发高性能API的首选。在待办事项管理场景中，用户需要通过API实现任务创建、读取、更新和删除等操作，这对路由设计的清晰性和数据验证的严谨性提出了高要求。

以一个典型待办事项应用为例，用户可能通过移动端或Web前端提交任务数据，后端API需确保数据完整性（如必填字段、日期格式）并返回标准化响应。FastAPI的Body验证和Path参数解析功能可高效处理这类需求，同时其内置的ASGI服务器支持高并发请求。

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

1. 环境搭建

使用Python 3.8+环境，通过pip install fastapi uvicorn安装核心依赖。推荐创建虚拟环境隔离项目依赖：

python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate     # Windows

2. 项目结构

采用分层架构设计：

todo_api/
├── main.py          # 入口文件
├── models.py        # 数据模型
├── schemas.py       # 请求/响应模型
├── crud.py          # 数据操作逻辑
└── tests/           # 测试用例

3. 基础路由设置

在main.py中初始化FastAPI应用，并添加根路由测试接口：

from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
    return {"message": "待办事项API服务已启动"}

三、数据模型与请求验证

1. Pydantic模型定义

在schemas.py中定义任务数据的请求/响应模型，利用Pydantic的验证机制确保数据合法性：

from pydantic import BaseModel, Field
from datetime import date
class TodoCreate(BaseModel):
    title: str = Field(..., min_length=3, max_length=50)
    description: str | None = Field(None, max_length=200)
    due_date: date | None = None
    priority: int = Field(1, ge=1, le=5)  # 1-5级优先级
class TodoUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    due_date: date | None = None
    priority: int | None = None
    is_completed: bool | None = None
class TodoResponse(TodoCreate):
    id: int
    is_completed: bool = False

2. 内存数据存储

为简化演示，使用字典模拟数据库存储（实际项目可替换为SQLAlchemy或Tortoise-ORM）：

# crud.py
from typing import Dict, List
from schemas import TodoCreate, TodoUpdate, TodoResponse
db: Dict[int, TodoResponse] = {}
next_id = 1
def get_todos() -> List[TodoResponse]:
    return list(db.values())
def create_todo(todo: TodoCreate) -> TodoResponse:
    global next_id
    new_todo = TodoResponse(id=next_id, **todo.dict())
    db[next_id] = new_todo
    next_id += 1
    return new_todo
# 其他CRUD函数...

四、CRUD路由实现详解

1. 创建任务（POST /todos/）

from fastapi import APIRouter, HTTPException
from schemas import TodoCreate, TodoResponse
import crud
router = APIRouter()
@router.post("/", response_model=TodoResponse)
async def create_todo(todo: TodoCreate):
    try:
        return crud.create_todo(todo)
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

关键点：

• 使用response_model自动序列化响应
• 通过HTTPException处理业务异常
• 请求体自动由Pydantic模型验证

2. 查询任务（GET /todos/ 与 GET /todos/{id}）

@router.get("/")
async def read_todos():
    return {"todos": crud.get_todos()}
@router.get("/{todo_id}", response_model=TodoResponse)
async def read_todo(todo_id: int):
    if todo_id not in crud.db:
        raise HTTPException(status_code=404, detail="任务未找到")
    return crud.db[todo_id]

优化实践：

• 添加分页参数：limit: int = 10, offset: int = 0
• 支持按状态过滤：completed: bool | None = None

3. 更新任务（PUT /todos/{id}）

@router.put("/{todo_id}", response_model=TodoResponse)
async def update_todo(todo_id: int, todo_update: TodoUpdate):
    if todo_id not in crud.db:
        raise HTTPException(404)
    todo = crud.db[todo_id]
    update_data = todo_update.dict(exclude_unset=True)
    updated_todo = todo.copy(update=update_data)
    crud.db[todo_id] = updated_todo
    return updated_todo

技术要点：

• exclude_unset=True避免覆盖未修改字段
• 使用字典的copy(update=)方法实现部分更新

4. 删除任务（DELETE /todos/{id}）

@router.delete("/{todo_id}")
async def delete_todo(todo_id: int):
    if todo_id not in crud.db:
        raise HTTPException(404)
    del crud.db[todo_id]
    return {"message": "任务删除成功"}

安全建议：

• 添加权限验证中间件
• 软删除替代物理删除（标记is_deleted字段）

五、路由整合与启动服务

在main.py中整合路由并启动服务：

from fastapi import FastAPI
from todo_router import router as todo_router
app = FastAPI()
app.include_router(todo_router, prefix="/todos", tags=["待办事项"])
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

启动参数说明：

• --reload：开发模式自动重载
• --workers 4：生产环境多进程配置

六、API测试与文档验证

1. 自动生成OpenAPI文档

访问http://localhost:8000/docs，Swagger UI提供交互式测试界面：

2. 使用HTTPie测试

# 创建任务
http POST :8000/todos/ title="学习FastAPI" priority=3
# 查询所有任务
http GET :8000/todos/
# 更新任务
http PUT :8000/todos/1 is_completed=true

3. 单元测试示例

在tests/test_todo.py中编写pytest用例：

from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_create_todo():
    response = client.post("/todos/", json={
        "title": "测试任务",
        "priority": 2
    })
    assert response.status_code == 200
    assert "id" in response.json()

七、进阶优化方向

1. 持久化存储：集成SQLite或PostgreSQL
2. 认证授权：添加JWT或OAuth2.0支持
3. 异步处理：使用async/await优化IO操作
4. 性能监控：集成Prometheus指标收集
5. API版本控制：通过路由前缀实现/v1/todos/

八、完整代码示例

# main.py 完整示例
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Optional
from datetime import date
app = FastAPI()
class Todo(BaseModel):
    title: str
    description: Optional[str] = None
    due_date: Optional[date] = None
    priority: int = 1
    is_completed: bool = False
db: Dict[int, Todo] = {}
next_id = 1
@app.post("/todos/")
async def create_todo(todo: Todo):
    global next_id
    todo_dict = todo.dict()
    todo_dict["id"] = next_id
    db[next_id] = Todo(**todo_dict)
    next_id += 1
    return todo_dict
@app.get("/todos/{todo_id}")
async def get_todo(todo_id: int):
    if todo_id not in db:
        raise HTTPException(404, "任务不存在")
    return db[todo_id]
# 其他路由实现...

通过以上步骤，开发者可快速构建一个功能完整的待办事项API服务。FastAPI的强类型支持和自动化工具链能显著提升开发效率，建议结合实际业务需求逐步扩展功能模块。