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

一、FastAPI框架核心优势与待办事项API开发价值

FastAPI作为基于Python的现代Web框架，凭借其异步支持、自动文档生成和类型注解特性，成为构建高性能API的首选方案。待办事项管理作为典型的CRUD应用场景，能够充分展现FastAPI在路由设计、数据验证和请求处理方面的优势。通过实现完整的增删改查功能，开发者可以掌握FastAPI的核心开发模式，为构建复杂业务系统奠定基础。

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

1. 环境准备

# 创建虚拟环境并安装依赖
python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate     # Windows
pip install fastapi uvicorn pydantic sqlalchemy

2. 项目结构规划

todo_api/
├── main.py            # 主入口文件
├── models.py          # 数据模型定义
├── schemas.py         # 数据验证Schema
├── crud.py            # 数据操作逻辑
└── database.py        # 数据库连接

三、核心组件实现详解

1. 数据模型定义（models.py）

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime
from .database import Base
class TodoItem(Base):
    __tablename__ = "todo_items"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), nullable=False)
    description = Column(String(500))
    completed = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

2. 数据验证Schema（schemas.py）

from pydantic import BaseModel
from datetime import datetime
class TodoItemCreate(BaseModel):
    title: str
    description: str = None
    completed: bool = False
class TodoItemUpdate(BaseModel):
    title: str = None
    description: str = None
    completed: bool = None
class TodoItem(BaseModel):
    id: int
    title: str
    description: str = None
    completed: bool
    created_at: datetime
    updated_at: datetime
    class Config:
        orm_mode = True

3. 数据库操作层（crud.py）

from sqlalchemy.orm import Session
from . import models, schemas
def get_todo(db: Session, todo_id: int):
    return db.query(models.TodoItem).filter(models.TodoItem.id == todo_id).first()
def get_todos(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.TodoItem).offset(skip).limit(limit).all()
def create_todo(db: Session, todo: schemas.TodoItemCreate):
    db_todo = models.TodoItem(
        title=todo.title,
        description=todo.description,
        completed=todo.completed
    )
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return db_todo
def update_todo(db: Session, todo_id: int, todo: schemas.TodoItemUpdate):
    db_todo = get_todo(db, todo_id)
    if not db_todo:
        return None
    update_data = todo.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
def delete_todo(db: Session, todo_id: int):
    db_todo = get_todo(db, todo_id)
    if db_todo:
        db.delete(db_todo)
        db.commit()
    return db_todo

四、路由设计与实现（main.py）

1. 依赖注入配置

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

2. CRUD路由实现

# 创建待办事项
@app.post("/todos/", response_model=schemas.TodoItem)
def create_todo(
    todo: schemas.TodoItemCreate,
    db: Session = Depends(get_db)
):
    return crud.create_todo(db=db, todo=todo)
# 获取所有待办事项
@app.get("/todos/", response_model=list[schemas.TodoItem])
def read_todos(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    todos = crud.get_todos(db=db, skip=skip, limit=limit)
    return todos
# 获取单个待办事项
@app.get("/todos/{todo_id}", response_model=schemas.TodoItem)
def read_todo(
    todo_id: int,
    db: Session = Depends(get_db)
):
    db_todo = crud.get_todo(db=db, todo_id=todo_id)
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return db_todo
# 更新待办事项
@app.put("/todos/{todo_id}", response_model=schemas.TodoItem)
def update_todo(
    todo_id: int,
    todo: schemas.TodoItemUpdate,
    db: Session = Depends(get_db)
):
    updated_todo = crud.update_todo(db=db, todo_id=todo_id, todo=todo)
    if updated_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return updated_todo
# 删除待办事项
@app.delete("/todos/{todo_id}")
def delete_todo(
    todo_id: int,
    db: Session = Depends(get_db)
):
    deleted_todo = crud.delete_todo(db=db, todo_id=todo_id)
    if deleted_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return {"message": "Todo deleted successfully"}

五、运行与测试

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 CRUD 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 '{"completed":true}'
# 删除待办事项
curl -X DELETE "http://127.0.0.1:8000/todos/1"

六、进阶优化建议

1. 分页实现：在get_todos路由中添加分页参数处理
2. 认证授权：集成OAuth2或JWT实现API安全
3. 异步支持：使用SQLAlchemy 2.0的异步API
4. 缓存机制：添加Redis缓存层提升性能
5. 日志记录：实现结构化日志记录

七、常见问题解决方案

1. 跨域问题：添加CORS中间件
   ```python
   from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
CORSMiddleware,
allow_origins=[““],
allow_credentials=True,
allow_methods=[““],
allow_headers=[“*”],
)

2. **数据库连接池**：配置合理的连接池参数
```python
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./todo.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    pool_size=5,
    max_overflow=10,
    pool_timeout=30
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

通过以上实现，开发者可以快速构建一个功能完整的待办事项Web API服务。FastAPI的自动文档功能（访问/docs）能够自动生成交互式API文档，极大提升开发效率。建议在实际项目中进一步考虑异常处理、输入验证和性能优化等高级特性。