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

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

一、FastAPI框架核心优势与项目初始化

FastAPI作为基于Python的现代Web框架,其核心优势体现在自动生成API文档高性能异步支持类型注解验证三大特性。相比Flask/Django,FastAPI通过Starlette和Pydantic的集成,使API开发效率提升40%以上(据FastAPI官方基准测试)。

项目初始化需完成三步:

  1. 环境配置:创建虚拟环境并安装核心依赖

    1. python -m venv venv
    2. source venv/bin/activate # Linux/Mac
    3. pip install fastapi uvicorn pydantic
  2. 项目结构规划:

    1. todo_api/
    2. ├── main.py # 路由入口
    3. ├── models.py # 数据模型
    4. ├── schemas.py # 请求/响应模型
    5. ├── crud.py # 数据操作层
    6. └── database.py # 模拟数据库
  3. 基础路由创建:
    ```python
    from fastapi import FastAPI

app = FastAPI()

@app.get(“/“)
def read_root():
return {“message”: “待办事项API服务”}

  1. ## 二、数据模型与请求验证设计
  2. ### 1. 数据模型定义(models.py)
  3. 使用Pydantic构建强类型数据模型:
  4. ```python
  5. from pydantic import BaseModel, Field
  6. from typing import Optional
  7. class Todo(BaseModel):
  8. id: Optional[int] = Field(default=None, description="唯一标识符")
  9. title: str = Field(..., min_length=3, max_length=50)
  10. description: Optional[str] = Field(default="", max_length=200)
  11. completed: bool = False

2. 请求/响应模型分离(schemas.py)

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

3. 内存数据库实现(database.py)

  1. from typing import Dict, List, Optional
  2. from models import Todo
  3. class TodoDatabase:
  4. def __init__(self):
  5. self.todos: Dict[int, Todo] = {}
  6. self.counter = 1
  7. def create(self, todo_data: dict) -> Todo:
  8. new_todo = Todo(id=self.counter, **todo_data)
  9. self.todos[self.counter] = new_todo
  10. self.counter += 1
  11. return new_todo
  12. def get_all(self) -> List[Todo]:
  13. return list(self.todos.values())
  14. # 其他CRUD方法实现...

三、路由设计与增删改查实现

1. 创建路由(POST /todos/)

  1. from fastapi import APIRouter, HTTPException
  2. from schemas import TodoCreate
  3. from database import TodoDatabase
  4. router = APIRouter()
  5. db = TodoDatabase()
  6. @router.post("/", response_model=Todo)
  7. def create_todo(todo: TodoCreate):
  8. try:
  9. return db.create(todo.dict())
  10. except Exception as e:
  11. raise HTTPException(status_code=400, detail=str(e))

2. 查询路由实现

  • 获取全部待办事项(GET /todos/)

    1. @router.get("/", response_model=List[Todo])
    2. def read_todos():
    3. return db.get_all()
  • 获取单个待办事项(GET /todos/{todo_id})

    1. @router.get("/{todo_id}", response_model=Todo)
    2. def read_todo(todo_id: int):
    3. if todo_id not in db.todos:
    4. raise HTTPException(status_code=404, detail="待办事项不存在")
    5. return db.todos[todo_id]

3. 更新路由实现(PUT /todos/{todo_id})

  1. from schemas import TodoUpdate
  2. @router.put("/{todo_id}", response_model=Todo)
  3. def update_todo(todo_id: int, todo_update: TodoUpdate):
  4. if todo_id not in db.todos:
  5. raise HTTPException(status_code=404)
  6. stored_todo = db.todos[todo_id]
  7. update_data = todo_update.dict(exclude_unset=True)
  8. updated_todo = stored_todo.copy(update=update_data)
  9. db.todos[todo_id] = updated_todo
  10. return updated_todo

4. 删除路由实现(DELETE /todos/{todo_id})

  1. @router.delete("/{todo_id}", response_model=dict)
  2. def delete_todo(todo_id: int):
  3. if todo_id not in db.todos:
  4. raise HTTPException(status_code=404)
  5. del db.todos[todo_id]
  6. return {"message": "待办事项已删除"}

四、项目整合与运行

1. 路由整合(main.py)

  1. from fastapi import FastAPI
  2. from routers import todo_router # 假设路由已模块化
  3. app = FastAPI()
  4. app.include_router(todo_router.router, prefix="/todos", tags=["待办事项"])

2. 启动服务

  1. uvicorn main:app --reload --host 0.0.0.0 --port 8000

3. 测试验证

使用curl测试创建待办事项:

  1. curl -X POST "http://localhost:8000/todos/" \
  2. -H "Content-Type: application/json" \
  3. -d '{"title": "学习FastAPI", "description": "完成API开发教程"}'

五、进阶优化建议

  1. 数据库集成:替换内存数据库为SQLite/PostgreSQL
    ```python

    使用SQLModel示例

    from sqlmodel import SQLModel, Field, Session, create_engine

class TodoDB(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
title: str

  1. # 其他字段...

engine = create_engine(“sqlite:///todos.db”)
SQLModel.metadata.create_all(engine)

  1. 2. **身份验证**:集成JWT认证
  2. ```python
  3. from fastapi.security import OAuth2PasswordBearer
  4. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  5. @app.get("/protected")
  6. async def protected_route(token: str = Depends(oauth2_scheme)):
  7. return {"message": "认证成功"}
  1. 性能优化
  • 启用异步路由处理
  • 实现请求缓存中间件
  • 添加API版本控制(如/v1/todos/)

六、常见问题解决方案

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

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

  1. 2. **请求体验证失败**:检查Pydantic模型字段约束
  2. ```python
  3. # 示例:修正字段定义
  4. class TodoCreate(BaseModel):
  5. title: str = Field(..., min_length=3, regex="^[A-Za-z0-9 ]+$")
  1. 路由冲突:确保路径前缀唯一性
    ```python

    错误示例(会导致冲突)

    app.include_router(router1, prefix=”/api”)
    app.include_router(router2, prefix=”/api”) # 冲突

正确做法

app.include_router(router1, prefix=”/api/v1”)
app.include_router(router2, prefix=”/api/v2”)

  1. ## 七、部署最佳实践
  2. 1. **Docker化部署**:
  3. ```dockerfile
  4. FROM python:3.9-slim
  5. WORKDIR /app
  6. COPY requirements.txt .
  7. RUN pip install --no-cache-dir -r requirements.txt
  8. COPY . .
  9. CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
  1. 生产环境配置
  • 禁用自动重载
  • 启用HTTPS
  • 配置适当的Worker数量
    1. gunicorn -k uvicorn.workers.UvicornWorker -w 4 -b :8000 main:app
  1. 监控与日志
  • 集成Prometheus指标
  • 配置结构化日志
    ```python
    from fastapi import Request
    from fastapi.logger import logger

@app.middleware(“http”)
async def log_requests(request: Request, call_next):
logger.info(f”请求路径: {request.url.path}”)
response = await call_next(request)
return response
```

通过以上完整实现,开发者可以快速构建一个符合RESTful规范的待办事项API服务。FastAPI的自动文档功能(访问/docs或/redoc)将显著提升前后端协作效率,而类型系统则能有效减少运行时错误。建议后续扩展方向包括:添加用户系统、实现任务分类、集成定时提醒功能等。