FastAPI 高级用法¶

pip install "fastapi[standard]"  # 安装 FastAPI 和所有推荐依赖
# 包含 uvicorn（ASGI 服务器）、pydantic、httpx 等

app/
├── main.py          # 入口
├── config.py        # 配置
├── models.py        # 数据模型
├── schemas.py       # Pydantic 模型
├── database.py      # 数据库连接
├── dependencies.py  # 共享依赖
├── routers/
│   ├── users.py     # 用户模块
│   └── items.py     # 物品模块
└── tests/
    └── test_main.py

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理（启动/关闭）"""
    print("启动：初始化数据库连接池")  # 启动时执行
    yield  # 应用运行中
    print("关闭：清理资源")  # 关闭时执行

app = FastAPI(
    title="生信数据 API",  # API 标题
    version="1.0.0",  # 版本号
    lifespan=lifespan,  # 生命周期管理
)

# 引入路由模块
from app.routers import users, items
app.include_router(users.router, prefix="/api/v1/users", tags=["用户"])
app.include_router(items.router, prefix="/api/v1/items", tags=["物品"])

# schemas.py — Pydantic 数据模型
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

class UserCreate(BaseModel):
    """创建用户的请求体"""
    username: str = Field(..., min_length=3, max_length=50, description="用户名")
    email: EmailStr = Field(..., description="邮箱")
    password: str = Field(..., min_length=8, description="密码")

class UserResponse(BaseModel):
    """用户响应模型（不返回密码）"""
    id: int
    username: str
    email: str
    created_at: datetime

    model_config = {"from_attributes": True}  # 支持从 ORM 对象转换

class PaginatedResponse(BaseModel):
    """分页响应"""
    total: int
    page: int
    items: list[UserResponse]

# routers/users.py — 用户路由
from fastapi import APIRouter, Depends, HTTPException, Query
from app.schemas import UserCreate, UserResponse, PaginatedResponse
from app.dependencies import get_db, get_current_user

router = APIRouter()

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db=Depends(get_db)):
    """创建新用户"""
    existing = await db.get_user_by_email(user.email)  # 检查邮箱是否已注册
    if existing:
        raise HTTPException(status_code=400, detail="邮箱已被注册")
    return await db.create_user(user)

@router.get("/", response_model=PaginatedResponse)
async def list_users(
    page: int = Query(1, ge=1, description="页码"),  # 查询参数，最小为1
    size: int = Query(10, ge=1, le=100, description="每页数量"),
    db=Depends(get_db),
):
    """获取用户列表（分页）"""
    total = await db.count_users()
    users = await db.get_users(skip=(page - 1) * size, limit=size)
    return PaginatedResponse(total=total, page=page, items=users)

@router.get("/me", response_model=UserResponse)
async def get_me(current_user=Depends(get_current_user)):
    """获取当前登录用户信息"""
    return current_user

# dependencies.py
from fastapi import Depends, HTTPException, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

async def get_db():
    """数据库会话依赖（每个请求一个连接，结束自动关闭）"""
    db = SessionLocal()
    try:
        yield db  # yield 让 FastAPI 自动管理生命周期
    finally:
        db.close()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db=Depends(get_db),
):
    """获取当前用户（从 JWT Token 解析）"""
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=["HS256"])
        user = await db.get_user(payload["sub"])
        if not user:
            raise HTTPException(status_code=401, detail="用户不存在")
        return user
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token 已过期")

from fastapi import BackgroundTasks

@router.post("/analyze")
async def start_analysis(
    sample_id: str,
    background_tasks: BackgroundTasks,  # 注入后台任务
):
    """提交分析任务（立即返回，后台异步执行）"""
    background_tasks.add_task(run_pipeline, sample_id)  # 添加后台任务
    return {"message": "分析已提交", "sample_id": sample_id}

async def run_pipeline(sample_id: str):
    """实际的分析流程（在后台运行）"""
    # 耗时的生信分析...
    pass

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await websocket.accept()  # 接受连接
    try:
        while True:
            data = await websocket.receive_text()  # 接收消息
            await websocket.send_text(f"收到: {data}")  # 发送消息
    except WebSocketDisconnect:
        print(f"客户端 {client_id} 断开连接")

from fastapi.middleware.cors import CORSMiddleware
import time

# CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 允许的前端域名
    allow_methods=["*"],  # 允许所有 HTTP 方法
    allow_headers=["*"],  # 允许所有请求头
)

# 自定义中间件：请求计时
@app.middleware("http")
async def add_timing(request, call_next):
    start = time.time()  # 请求开始时间
    response = await call_next(request)  # 处理请求
    duration = time.time() - start  # 计算耗时
    response.headers["X-Process-Time"] = str(duration)  # 添加到响应头
    return response

# === 启动命令 ===
uvicorn app.main:app --reload          # 开发模式（自动重载）
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4  # 生产模式

# === 自动文档 ===
# Swagger UI:  http://localhost:8000/docs
# ReDoc:       http://localhost:8000/redoc
# OpenAPI JSON: http://localhost:8000/openapi.json

# === 常用装饰器 ===
@app.get("/path")                      # GET 请求
@app.post("/path", status_code=201)    # POST 请求
@app.put("/path")                      # PUT 请求
@app.delete("/path", status_code=204)  # DELETE 请求

# === 参数类型 ===
# 路径参数: /users/{user_id}
# 查询参数: ?page=1&size=10
# 请求体:   Pydantic Model
# 请求头:   Header(...)
# Cookie:   Cookie(...)
# 文件上传: UploadFile

# === 依赖注入 ===
Depends(function)                      # 注入依赖