FastAPI 模式

FastAPI 模式

概览

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于使用 Python 3.6+ 构建 API，使用标准 Python 类型提示，使开发者能够快速、可维护且类型安全地创建 API。

FastAPI 包括：

• 类型提示：标准 Python 类型提示用于自动验证
• Pydantic：使用 Pydantic 模型进行数据验证
• 异步支持：原生异步/等待支持
• OpenAPI：自动 OpenAPI 文档
• 依赖注入：强大的依赖注入系统
• WebSocket 支持：内置 WebSocket 支持

为什么这很重要

• 提高开发速度：FastAPI 可以提高开发速度 40-60%
• 减少错误：自动验证可以减少错误 30-40%
• 增强性能：高性能框架增强 API 性能
• 减少文档时间：自动生成的文档减少文档时间
• 提高类型安全：类型提示提高类型安全

---

核心概念

1. Pydantic 模型

from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    password: str = Field(..., min_length=8, max_length=100)

    @validator("password")
    def validate_password(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError("密码必须包含大写字母")
        if not any(c.islower() for c in v):
            raise ValueError("密码必须包含小写字母")
        if not any(c.isdigit() for c in v):
            raise ValueError("密码必须包含数字")
        return v

class UserResponse(UserBase):
    id: int
    is_active: bool = True
    created_at: datetime

2. 依赖注入

from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt

async def get_current_user(
    db: Session = Depends(get_db),
    token: str = Depends(oauth2_scheme)
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
    )

    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        token_data = TokenPayload(**payload)
    except JWTError:
        raise credentials_exception

    user = db.query(User).filter(User.id == token_data.sub).first()
    if not user:
        raise credentials_exception

    return user

3. 异步数据库操作

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

async_engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db"
)
AsyncSessionLocal = async_sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

async def get_async_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

# 异步端点
from fastapi import Depends

@app.get("/users")
async def get_users_async(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(User))
    users = result.scalars().all()
    return users

4. 错误处理

from fastapi import HTTPException, status

class NotFoundException(HTTPException):
    def __init__(self, detail: str = "资源未找到"):
        super().__init__(status_code=status.HTTP_404_NOT_FOUND, detail=detail)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise NotFoundException()
    return user

5. 后台任务

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # 发送邮件逻辑
    print(f"发送给 {email}: {message}")

@app.post("/send-email")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_email, email, "欢迎!")
    return {"message": "邮件将在后台发送"}

6. WebSocket

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"客户端 {client_id}: {data}")
    except WebSocketDisconnect:
        print(f"客户端 {client_id} 断开连接")

7. 文件上传

from fastapi import UploadFile, File
import shutil

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    file_location = f"uploads/{file.filename}"
    with open(file_location, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"filename": file.filename, "location": file_location}

快速开始

1. 安装 FastAPI：

   pip install fastapi uvicorn
   

2. 创建基本应用：

   from fastapi import FastAPI
   
   app = FastAPI()
   
   @app.get("/")
   async def root():
       return {"message": "Hello World"}
   

3. 运行服务器：

   uvicorn main:app --reload
   

4. 查看文档：

   • Swagger UI: http://localhost:8000/docs
   • ReDoc: http://localhost:8000/redoc

生产检查清单

• [ ] 使用 Python 类型提示进行类型安全
• [ ] 实现 Pydantic 模型进行验证
• [ ] 使用异步/等待进行异步操作
• [ ] 实施依赖注入
• [ ] 添加认证和授权
• [ ] 使用中间件处理跨领域问题
• [ ] 实施适当的错误处理
• [ ] 添加后台任务进行异步操作
• [ ] 使用 Pytest 测试端点
• [ ] 使用 OpenAPI 文档
• [ ] 添加 CORS 配置
• [ ] 实施速率限制
• [ ] 使用 WebSocket 进行实时功能
• [ ] 添加文件上传支持
• [ ] 监控性能和错误

反模式

1. 类型提示不当：不要错误使用类型提示
2. 缺少验证：不要跳过输入验证
3. 错误处理不当：不要不恰当地处理错误
4. 缺少文档：不要未能为 API 文档
5. 性能不佳：不要未能优化性能
6. 缺少测试：不要未能测试端点

集成点

• 错误处理：03-backend-api/error-handling
• 验证：03-backend-api/validation
• 中间件：03-backend-api/middleware
• Python 标准：01-foundations/python-standards
• 监控：14-monitoring-observability

进一步阅读

• FastAPI 文档
• FastAPI 教程
• Pydantic 文档