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

2025年10月17日 互联网

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

在当今快速发展的软件开发领域，Web API作为连接前端与后端、实现数据交互的关键桥梁，其重要性不言而喻。FastAPI，作为一款基于Python的现代、快速（高性能）的Web框架，凭借其自动生成API文档、支持异步请求处理以及类型提示等特性，迅速成为开发者构建高效Web API的首选工具之一。本文将围绕“FastAPI快速开发Web API项目：实现FastAPI的待办事项路由增删改查”这一主题，详细阐述如何利用FastAPI框架快速搭建一个具备增删改查（CRUD）功能的待办事项Web API项目。

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

1.1 环境准备

在开始之前，确保你的开发环境中已安装Python 3.7+版本。FastAPI推荐使用Uvicorn作为ASGI服务器来运行应用，因此还需安装Uvicorn。此外，为了简化数据库操作，我们将使用SQLAlchemy作为ORM工具，以及Pydantic进行数据验证。

1.2 项目初始化

首先，创建一个新的Python虚拟环境，并安装FastAPI、Uvicorn、SQLAlchemy和Pydantic等依赖库：

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

1.3 项目结构规划

一个良好的项目结构有助于代码的维护与扩展。建议采用如下结构：

  1. todo_api/
  2. │── main.py            # 应用入口
  3. │── models.py          # 数据库模型
  4. │── schemas.py         # Pydantic模型
  5. │── crud.py            # 数据库操作
  6. │── database.py        # 数据库连接与会话管理
  7. │── routers/
  8. │   └── todos.py       # 待办事项路由
  9. └── requirements.txt   # 依赖列表

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

2.1 数据库模型（models.py）

使用SQLAlchemy定义待办事项的数据库模型，包括ID、标题、描述、完成状态等字段：

  1. from sqlalchemy import Column, Integer, String, Boolean
  2. from sqlalchemy.ext.declarative import declarative_base
  4. Base = declarative_base()
  6. class Todo(Base):
  7.     __tablename__ = 'todos'
  9.     id = Column(Integer, primary_key=True, index=True)
  10.     title = Column(String, index=True)
  11.     description = Column(String)
  12.     completed = Column(Boolean, default=False)

2.2 Pydantic模型（schemas.py）

定义Pydantic模型用于数据验证和序列化，包括待办事项的创建、更新和响应模型：

  1. from pydantic import BaseModel
  3. class TodoBase(BaseModel):
  4.     title: str
  5.     description: str = None
  6.     completed: bool = False
  8. class TodoCreate(TodoBase):
  9.     pass
  11. class TodoUpdate(TodoBase):
  12.     pass
  14. class Todo(TodoBase):
  15.     id: int
  17.     class Config:
  18.         orm_mode = True

三、数据库操作（crud.py）

实现待办事项的增删改查功能，包括创建、读取、更新和删除操作：

  1. from sqlalchemy.orm import Session
  2. from . import models, schemas
  4. def get_todo(db: Session, todo_id: int):
  5.     return db.query(models.Todo).filter(models.Todo.id == todo_id).first()
  7. def get_todos(db: Session, skip: int = 0, limit: int = 100):
  8.     return db.query(models.Todo).offset(skip).limit(limit).all()
  10. def create_todo(db: Session, todo: schemas.TodoCreate):
  11.     db_todo = models.Todo(**todo.dict())
  12.     db.add(db_todo)
  13.     db.commit()
  14.     db.refresh(db_todo)
  15.     return db_todo
  17. def update_todo(db: Session, todo_id: int, todo: schemas.TodoUpdate):
  18.     db_todo = get_todo(db, todo_id)
  19.     if db_todo:
  20.         for key, value in todo.dict(exclude_unset=True).items():
  21.             setattr(db_todo, key, value)
  22.         db.commit()
  23.         db.refresh(db_todo)
  24.     return db_todo
  26. def delete_todo(db: Session, todo_id: int):
  27.     db_todo = get_todo(db, todo_id)
  28.     if db_todo:
  29.         db.delete(db_todo)
  30.         db.commit()
  31.     return db_todo

四、路由定义与API实现（routers/todos.py）

定义待办事项的路由，包括获取所有待办事项、获取单个待办事项、创建待办事项、更新待办事项和删除待办事项：

  1. from fastapi import APIRouter, Depends, HTTPException
  2. from sqlalchemy.orm import Session
  3. from .. import crud, models, schemas
  4. from ..database import get_db
  6. router = APIRouter()
  8. @router.get('/todos/', response_model=list[schemas.Todo])
  9. def read_todos(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
  10.     todos = crud.get_todos(db, skip=skip, limit=limit)
  11.     return todos
  13. @router.get('/todos/{todo_id}', response_model=schemas.Todo)
  14. def read_todo(todo_id: int, db: Session = Depends(get_db)):
  15.     db_todo = crud.get_todo(db, todo_id)
  16.     if db_todo is None:
  17.         raise HTTPException(status_code=404, detail='Todo not found')
  18.     return db_todo
  20. @router.post('/todos/', response_model=schemas.Todo)
  21. def create_todo(todo: schemas.TodoCreate, db: Session = Depends(get_db)):
  22.     return crud.create_todo(db, todo)
  24. @router.put('/todos/{todo_id}', response_model=schemas.Todo)
  25. def update_todo(todo_id: int, todo: schemas.TodoUpdate, db: Session = Depends(get_db)):
  26.     db_todo = crud.update_todo(db, todo_id, todo)
  27.     if db_todo is None:
  28.         raise HTTPException(status_code=404, detail='Todo not found')
  29.     return db_todo
  31. @router.delete('/todos/{todo_id}')
  32. def delete_todo(todo_id: int, db: Session = Depends(get_db)):
  33.     db_todo = crud.delete_todo(db, todo_id)
  34.     if db_todo is None:
  35.         raise HTTPException(status_code=404, detail='Todo not found')
  36.     return {'message': 'Todo deleted successfully'}

五、应用入口与数据库连接（main.py & database.py）

5.1 数据库连接（database.py）

  1. from sqlalchemy import create_engine
  2. from sqlalchemy.orm import sessionmaker
  4. SQLALCHEMY_DATABASE_URL = 'sqlite:///./todo.db'  # 使用SQLite数据库
  6. engine = create_engine(
  7.     SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False}
  8. )
  9. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
  11. def get_db():
  12.     db = SessionLocal()
  13.     try:
  14.         yield db
  15.     finally:
  16.         db.close()

5.2 应用入口（main.py）

  1. from fastapi import FastAPI
  2. from .routers import todos
  3. from .database import Base, engine
  5. # 创建数据库表
  6. Base.metadata.create_all(bind=engine)
  8. app = FastAPI()
  10. app.include_router(todos.router)

六、运行与测试

使用Uvicorn运行应用：

  1. uvicorn main:app --reload

应用启动后，可通过访问http://127.0.0.1:8000/docs查看自动生成的API文档，并进行交互式测试。

七、总结与展望

通过本文的介绍，我们成功利用FastAPI框架快速搭建了一个具备增删改查功能的待办事项Web API项目。FastAPI的高性能、自动生成API文档以及类型提示等特性，极大地提高了开发效率与代码质量。未来，你可以进一步扩展项目，如添加用户认证、权限控制、数据分页、更复杂的查询逻辑等，以满足更丰富的业务需求。

最新文章