部署与性能优化

pydantic-settings 环境变量管理

# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    # 自动从环境变量读取（优先级：环境变量 > .env 文件 > 默认值）
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False
    )

    # 数据库
    database_url: str = "sqlite+aiosqlite:///./app.db"

    # JWT
    secret_key: str = "change-me-in-production"
    access_token_expire_minutes: int = 30

    # Redis
    redis_url: str = "redis://localhost:6379"

    # 应用配置
    debug: bool = False
    allowed_hosts: list[str] = ["*"]
    api_v1_prefix: str = "/api/v1"

    # 第三方 API Key（从环境变量读取，不设默认值）
    anthropic_api_key: str = ""
    openai_api_key: str = ""

@lru_cache()
def get_settings() -> Settings:
    """缓存 Settings 实例，避免重复读取文件"""
    return Settings()

settings = get_settings()

Gunicorn + Uvicorn 生产配置

# gunicorn.conf.py
import multiprocessing

# Worker 配置：CPU 核心数 × 2 + 1（常用公式）
workers = multiprocessing.cpu_count() * 2 + 1

# 使用 UvicornWorker 以支持 ASGI
worker_class = "uvicorn.workers.UvicornWorker"

# 绑定地址
bind = "0.0.0.0:8000"

# 超时设置
timeout = 120              # 请求超时（秒），AI 接口可适当延长
keepalive = 5              # 连接保持活跃时间
graceful_timeout = 30      # 优雅关闭等待时间

# 进程管理
max_requests = 1000        # 每个 worker 处理 1000 个请求后重启（防内存泄漏）
max_requests_jitter = 100  # 抖动，避免所有 worker 同时重启

# 日志
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s"'

# 生产启动命令
gunicorn app.main:app -c gunicorn.conf.py

# 或使用 uvicorn（单进程，适合容器环境）
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

Docker 多阶段构建

# Dockerfile
# ── 构建阶段 ──────────────────────────────────────────────
FROM python:3.12-slim AS builder

WORKDIR /app

# 安装 uv（比 pip 快 10-100x）
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv

# 先复制依赖文件（利用 Docker 层缓存）
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project

# ── 生产阶段（最小化镜像）────────────────────────────────
FROM python:3.12-slim AS production

WORKDIR /app

# 从构建阶段复制虚拟环境
COPY --from=builder /app/.venv /app/.venv

# 复制应用代码
COPY app/ ./app/
COPY gunicorn.conf.py ./

# 非 root 用户运行（安全最佳实践）
RUN useradd --no-create-home --shell /bin/false appuser
USER appuser

ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --start-period=10s \
  CMD python -c "import httpx; httpx.get('http://localhost:8000/health')"

CMD ["gunicorn", "app.main:app", "-c", "gunicorn.conf.py"]

# docker-compose.yml（开发环境）
version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/mydb
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine

volumes:
  pgdata:

Redis 缓存

from fastapi import FastAPI, Depends
import redis.asyncio as redis
import json

app = FastAPI()
redis_client = redis.from_url("redis://localhost:6379", decode_responses=True)

# ── 手动缓存模式 ──────────────────────────────────────────
@app.get("/products/{product_id}")
async def get_product(product_id: int):
    cache_key = f"product:{product_id}"

    # 1. 查 Redis 缓存
    cached = await redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # 2. 缓存未命中，查数据库
    product = {"id": product_id, "name": "示例商品", "price": 99.9}

    # 3. 写入 Redis，TTL 300 秒
    await redis_client.setex(cache_key, 300, json.dumps(product))
    return product

# ── 更新时清除缓存 ────────────────────────────────────────
@app.put("/products/{product_id}")
async def update_product(product_id: int):
    # 更新数据库（省略）
    # 删除缓存（Cache Invalidation）
    await redis_client.delete(f"product:{product_id}")
    return {"updated": product_id}

API 限流（slowapi）

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

# 基于客户端 IP 限流（生产环境建议使用 Redis 存储计数）
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# ── 限流注解 ──────────────────────────────────────────────
@app.get("/api/search")
@limiter.limit("30/minute")   # 每 IP 每分钟 30 次
async def search(request: Request, q: str):
    return {"results": []}

@app.post("/api/ai/chat")
@limiter.limit("10/minute;100/hour")  # 组合限流规则
async def ai_chat(request: Request, message: str):
    return {"reply": "..."}