FastAPI实战:待办事项API的增删改查全流程实现
一、FastAPI框架核心优势与项目初始化
FastAPI作为基于Python的现代Web框架,其核心优势体现在自动生成API文档、高性能异步支持和类型注解验证三大特性。相比Flask/Django,FastAPI通过Starlette和Pydantic的集成,使API开发效率提升40%以上(据FastAPI官方基准测试)。
项目初始化需完成三步:
-
环境配置:创建虚拟环境并安装核心依赖
python -m venv venvsource venv/bin/activate # Linux/Macpip install fastapi uvicorn pydantic
-
项目结构规划:
todo_api/├── main.py # 路由入口├── models.py # 数据模型├── schemas.py # 请求/响应模型├── crud.py # 数据操作层└── database.py # 模拟数据库
-
基础路由创建:
```python
from fastapi import FastAPI
app = FastAPI()
@app.get(“/“)
def read_root():
return {“message”: “待办事项API服务”}
## 二、数据模型与请求验证设计### 1. 数据模型定义(models.py)使用Pydantic构建强类型数据模型:```pythonfrom pydantic import BaseModel, Fieldfrom typing import Optionalclass Todo(BaseModel):id: Optional[int] = Field(default=None, description="唯一标识符")title: str = Field(..., min_length=3, max_length=50)description: Optional[str] = Field(default="", max_length=200)completed: bool = False
2. 请求/响应模型分离(schemas.py)
from pydantic import BaseModelclass TodoCreate(BaseModel):title: strdescription: str = ""class TodoUpdate(BaseModel):title: Optional[str] = Nonedescription: Optional[str] = Nonecompleted: Optional[bool] = None
3. 内存数据库实现(database.py)
from typing import Dict, List, Optionalfrom models import Todoclass TodoDatabase:def __init__(self):self.todos: Dict[int, Todo] = {}self.counter = 1def create(self, todo_data: dict) -> Todo:new_todo = Todo(id=self.counter, **todo_data)self.todos[self.counter] = new_todoself.counter += 1return new_tododef get_all(self) -> List[Todo]:return list(self.todos.values())# 其他CRUD方法实现...
三、路由设计与增删改查实现
1. 创建路由(POST /todos/)
from fastapi import APIRouter, HTTPExceptionfrom schemas import TodoCreatefrom database import TodoDatabaserouter = APIRouter()db = TodoDatabase()@router.post("/", response_model=Todo)def create_todo(todo: TodoCreate):try:return db.create(todo.dict())except Exception as e:raise HTTPException(status_code=400, detail=str(e))
2. 查询路由实现
-
获取全部待办事项(GET /todos/)
@router.get("/", response_model=List[Todo])def read_todos():return db.get_all()
-
获取单个待办事项(GET /todos/{todo_id})
@router.get("/{todo_id}", response_model=Todo)def read_todo(todo_id: int):if todo_id not in db.todos:raise HTTPException(status_code=404, detail="待办事项不存在")return db.todos[todo_id]
3. 更新路由实现(PUT /todos/{todo_id})
from schemas import TodoUpdate@router.put("/{todo_id}", response_model=Todo)def update_todo(todo_id: int, todo_update: TodoUpdate):if todo_id not in db.todos:raise HTTPException(status_code=404)stored_todo = db.todos[todo_id]update_data = todo_update.dict(exclude_unset=True)updated_todo = stored_todo.copy(update=update_data)db.todos[todo_id] = updated_todoreturn updated_todo
4. 删除路由实现(DELETE /todos/{todo_id})
@router.delete("/{todo_id}", response_model=dict)def delete_todo(todo_id: int):if todo_id not in db.todos:raise HTTPException(status_code=404)del db.todos[todo_id]return {"message": "待办事项已删除"}
四、项目整合与运行
1. 路由整合(main.py)
from fastapi import FastAPIfrom routers import todo_router # 假设路由已模块化app = FastAPI()app.include_router(todo_router.router, prefix="/todos", tags=["待办事项"])
2. 启动服务
uvicorn main:app --reload --host 0.0.0.0 --port 8000
3. 测试验证
使用curl测试创建待办事项:
curl -X POST "http://localhost:8000/todos/" \-H "Content-Type: application/json" \-d '{"title": "学习FastAPI", "description": "完成API开发教程"}'
五、进阶优化建议
- 数据库集成:替换内存数据库为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
# 其他字段...
engine = create_engine(“sqlite:///todos.db”)
SQLModel.metadata.create_all(engine)
2. **身份验证**:集成JWT认证```pythonfrom fastapi.security import OAuth2PasswordBeareroauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")@app.get("/protected")async def protected_route(token: str = Depends(oauth2_scheme)):return {"message": "认证成功"}
- 性能优化:
- 启用异步路由处理
- 实现请求缓存中间件
- 添加API版本控制(如/v1/todos/)
六、常见问题解决方案
- 跨域问题:添加CORS中间件
```python
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=[““],
allow_methods=[““],
allow_headers=[“*”],
)
2. **请求体验证失败**:检查Pydantic模型字段约束```python# 示例:修正字段定义class TodoCreate(BaseModel):title: str = Field(..., min_length=3, regex="^[A-Za-z0-9 ]+$")
- 路由冲突:确保路径前缀唯一性
```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. **Docker化部署**:```dockerfileFROM python:3.9-slimWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtCOPY . .CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
- 生产环境配置:
- 禁用自动重载
- 启用HTTPS
- 配置适当的Worker数量
gunicorn -k uvicorn.workers.UvicornWorker -w 4 -b :8000 main:app
- 监控与日志:
- 集成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)将显著提升前后端协作效率,而类型系统则能有效减少运行时错误。建议后续扩展方向包括:添加用户系统、实现任务分类、集成定时提醒功能等。