FastAPI精通：异步架构、微服务与生产部署

from contextlib import asynccontextmanager
from redis import asyncio as aioredis

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：初始化连接池
    redis_pool = aioredis.ConnectionPool.from_url(
        "redis://localhost:6379",
        max_connections=50,
        decode_responses=True,
    )
    app.state.redis = aioredis.Redis(connection_pool=redis_pool)
    app.state.http_session = httpx.AsyncClient(timeout=30)

    yield  # 应用运行期间

    # 关闭时：优雅释放资源
    await app.state.redis.close()
    await app.state.http_session.aclose()
    await redis_pool.disconnect()

app = FastAPI(lifespan=lifespan)

import asyncio
from fastapi.responses import StreamingResponse

async def generate_large_csv():
    """百万行级 CSV 流式输出，内存消耗恒定 < 1MB"""
    header = "id,name,email,created_at\n"
    yield header
    batch_size = 10000
    offset = 0
    while True:
        batch = await fetch_users_batch(offset, batch_size)
        if not batch:
            break
        rows = "\n".join(
            f"{u.id},{u.name},{u.email},{u.created_at.isoformat()}"
            for u in batch
        )
        yield rows + "\n"
        offset += batch_size
        # 让出事件循环，避免阻塞其他请求
        await asyncio.sleep(0)

@app.get("/users/export")
async def export_users():
    return StreamingResponse(
        generate_large_csv(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=users.csv"},
    )

from fastapi import WebSocket, WebSocketDisconnect, WebSocketException
from typing import Dict, Set
import json
import uuid

class ConnectionManager:
    def __init__(self):
        # 房间 -> 用户连接映射
        self.rooms: Dict[str, Dict[str, WebSocket]] = {}

    async def connect(self, room: str, ws: WebSocket):
        await ws.accept()
        if room not in self.rooms:
            self.rooms[room] = {}
        user_id = str(uuid.uuid4())
        self.rooms[room][user_id] = ws
        return user_id

    def disconnect(self, room: str, user_id: str):
        self.rooms.get(room, {}).pop(user_id, None)
        if not self.rooms.get(room):
            self.rooms.pop(room, None)

    async def broadcast(self, room: str, message: dict, exclude: str = None):
        for uid, ws in self.rooms.get(room, {}).items():
            if uid == exclude:
                continue
            try:
                await ws.send_json(message)
            except Exception:
                self.disconnect(room, uid)

manager = ConnectionManager()

@app.websocket("/ws/chat/{room_id}")
async def chat_endpoint(ws: WebSocket, room_id: str, token: str = Query(...)):
    # 认证校验
    user = await verify_token(token)
    if not user:
        await ws.close(code=4001, reason="Unauthorized")
        return

    user_id = await manager.connect(room_id, ws)
    await manager.broadcast(room_id, {
        "type": "system",
        "content": f"{user.name} 加入了房间"
    })

    try:
        while True:
            data = await ws.receive_text()
            msg = json.loads(data)
            if msg.get("type") == "ping":
                await ws.send_json({"type": "pong"})
                continue
            await manager.broadcast(room_id, {
                "type": "message",
                "user": user.name,
                "content": msg["content"],
                "timestamp": datetime.utcnow().isoformat(),
            }, exclude=user_id)
    except WebSocketDisconnect:
        manager.disconnect(room_id, user_id)
        await manager.broadcast(room_id, {
            "type": "system",
            "content": f"{user.name} 离开了房间"
        })

# gateway/app/main.py — 基于 FastAPI 的轻量级 API Gateway
from fastapi import FastAPI, Request, HTTPException
import httpx
import asyncio

app = FastAPI(title="API Gateway")

ROUTES = {
    "/api/users": "http://user-service:8001",
    "/api/orders": "http://order-service:8002",
    "/api/payments": "http://payment-service:8003",
}

TIMEOUT = httpx.Timeout(timeout=10.0)
client = httpx.AsyncClient(timeout=TIMEOUT)

@app.api_route("/api/{service_path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def gateway_handler(request: Request, service_path: str):
    # 路由匹配
    for prefix, target in ROUTES.items():
        if service_path.startswith(prefix.strip("/api/")):
            target_url = f"{target}/{service_path}"

            # 转发请求体与头部（过滤 hop-by-hop headers）
            headers = dict(request.headers)
            for h in ["host", "content-length", "transfer-encoding"]:
                headers.pop(h, None)

            try:
                resp = await client.request(
                    method=request.method,
                    url=target_url,
                    headers=headers,
                    content=await request.body(),
                    params=request.query_params,
                )
                return Response(
                    content=resp.content,
                    status_code=resp.status_code,
                    headers=dict(resp.headers),
                )
            except httpx.RequestError as e:
                raise HTTPException(status_code=502, detail=str(e))

    raise HTTPException(status_code=404, detail="Service not found")

import aio_pika
from aio_pika import ExchangeType, Message, DeliveryMode

class TaskDispatcher:
    def __init__(self, amqp_url: str = "amqp://guest:guest@rabbitmq/"):
        self.amqp_url = amqp_url
        self.connection: aio_pika.RobustConnection = None
        self.channel: aio_pika.Channel = None
        self.exchange: aio_pika.Exchange = None

    async def connect(self):
        self.connection = await aio_pika.connect_robust(self.amqp_url)
        self.channel = await self.connection.channel()
        # 定义死信队列用于失败重试
        dlx = await self.channel.declare_exchange("dlx", ExchangeType.DIRECT)
        dlq = await self.channel.declare_queue("dead_letter_queue", durable=True)
        await dlq.bind(dlx, "dead")
        # 主交换机
        self.exchange = await self.channel.declare_exchange(
            "tasks", ExchangeType.TOPIC, durable=True
        )

    async def dispatch(self, routing_key: str, payload: dict, delay: int = 0):
        message = Message(
            body=json.dumps(payload).encode(),
            delivery_mode=DeliveryMode.PERSISTENT,
            headers={"x-delay": str(delay)} if delay else {},
        )
        await self.exchange.publish(message, routing_key)

    async def consume(self, routing_keys: list[str], callback):
        queue = await self.channel.declare_queue(durable=True)
        for key in routing_keys:
            await queue.bind(self.exchange, key)
        await queue.consume(callback)

# FastAPI 集成示例
@asynccontextmanager
async def lifespan(app):
    dispatcher = TaskDispatcher()
    await dispatcher.connect()
    app.state.dispatcher = dispatcher
    yield
    await dispatcher.connection.close()

@app.post("/tasks/send-email")
async def send_email_task(payload: EmailPayload):
    await app.state.dispatcher.dispatch(
        "email.send",
        payload.model_dump(),
        delay=0  # 延迟队列可用于定时发送
    )
    return {"status": "queued", "task_id": payload.task_id}

# worker.py — 独立进程运行
async def on_email_task(message: aio_pika.IncomingMessage):
    async with message.process(ignore_processed=True):
        data = json.loads(message.body)
        try:
            await send_email_service(data)
            await message.ack()
        except Exception as e:
            logger.error(f"Failed to process: {e}")
            # 重试逻辑：3次失败后进入死信队列
            if message.header.retry_count and message.header.retry_count >= 3:
                await message.reject(requeue=False)
            else:
                await message.reject(requeue=True)

async def main():
    dispatcher = TaskDispatcher()
    await dispatcher.connect()
    await dispatcher.consume(["email.send", "email.batch"], on_email_task)
    await asyncio.Future()  # 保持运行

if __name__ == "__main__":
    asyncio.run(main())

// proto/user_service.proto
service UserService {
  rpc GetUser (GetUserRequest) returns (UserResponse);
  rpc BatchGetUsers (BatchGetUsersRequest) returns (stream UserResponse);
}

message GetUserRequest { string user_id = 1; }
message UserResponse {
  string user_id = 1;
  string name = 2;
  string email = 3;
  int32  status = 4;
}

# grpc_client.py — FastAPI 中调用 gRPC 服务的模式
import grpc
import user_service_pb2_grpc
import user_service_pb2

class UserGrpcClient:
    def __init__(self, endpoint: str = "user-service:50051"):
        self.endpoint = endpoint
        self._channel: grpc.aio.Channel = None
        self._stub: user_service_pb2_grpc.UserServiceStub = None

    async def __aenter__(self):
        self._channel = grpc.aio.insecure_channel(
            self.endpoint,
            options=[
                ("grpc.keepalive_time_ms", 30000),
                ("grpc.keepalive_timeout_ms", 10000),
                ("grpc.max_send_message_length", 50 * 1024 * 1024),
            ],
        )
        self._stub = user_service_pb2_grpc.UserServiceStub(self._channel)
        return self

    async def __aexit__(self, *args):
        await self._channel.close()

    async def get_user(self, user_id: str) -> dict:
        response = await self._stub.GetUser(
            user_service_pb2.GetUserRequest(user_id=user_id)
        )
        return {
            "user_id": response.user_id,
            "name": response.name,
            "email": response.email,
        }

    async def batch_get_users(self, user_ids: list[str]) -> list[dict]:
        """流式读取：支持十万级用户的批量查询"""
        result = []
        async for response in self._stub.BatchGetUsers(
            user_service_pb2.BatchGetUsersRequest(user_ids=user_ids)
        ):
            result.append({
                "user_id": response.user_id,
                "name": response.name,
            })
        return result

同步调用 + 强一致性 = gRPC / HTTP（适合：用户查询、订单校验）
异步编排 + 最终一致性 = 事件驱动（适合：通知推送、数据同步、审计日志）

混合模式：关键路径同步调用，非关键路径异步事件。
例如：下单时 gRPC 同步校验库存，下单成功后通过 Kafka 异步触发发票生成。

# Dockerfile — 多阶段构建
FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /install /usr/local
COPY . .

# 非 root 用户运行
RUN groupadd -r fastapi && useradd -r -g fastapi fastapi
USER fastapi

EXPOSE 8000

# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
  CMD curl --fail http://localhost:8000/health || exit 1

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# gunicorn.conf.py — 生产级配置
import multiprocessing

bind = "0.0.0.0:8000"
worker_class = "uvicorn.workers.UvicornWorker"
workers = multiprocessing.cpu_count() * 2 + 1
max_requests = 10000          # 防止内存泄漏，定期重启 worker
max_requests_jitter = 2000    # 错峰重启
timeout = 120
keepalive = 65                # 保持连接，配合 Nginx
graceful_timeout = 30         # 优雅关闭超时
accesslog = "/var/log/app/access.log"
errorlog = "/var/log/app/error.log"
loglevel = "info"

upstream fastapi_backend {
    least_conn;                              # 最少连接负载均衡
    server app:8000 max_fails=3 fail_timeout=30s;
    keepalive 32;                            # 长连接池
}

server {
    listen 80;
    server_name api.example.com;

    # 请求体限制
    client_max_body_size 50m;

    # 压缩
    gzip on;
    gzip_types application/json text/plain text/css text/javascript;

    location / {
        proxy_pass http://fastapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # WebSocket 支持
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";

        # 超时配置
        proxy_connect_timeout 60s;
        proxy_send_timeout    120s;
        proxy_read_timeout    120s;
    }

    location /health {
        access_log off;
        return 200 "ok";
    }
}

# 数据库连接池配置 —— 以 asyncpg 为例
from asyncpg import create_pool
from dataclasses import dataclass

@dataclass
class PoolConfig:
    min_size: int = 10
    max_size: int = 50
    max_inactive_connection_lifetime: float = 60.0
    command_timeout: int = 30

class OptimizedPool:
    def __init__(self, dsn: str, config: PoolConfig):
        self.dsn = dsn
        self.config = config
        self.pool = None

    async def init(self):
        self.pool = await create_pool(
            self.dsn,
            min_size=self.config.min_size,
            max_size=self.config.max_size,
            max_inactive_connection_lifetime=self.config.max_inactive_connection_lifetime,
            command_timeout=self.config.command_timeout,
            # 重要：禁用 statement cache 对于长时间运行服务可减少内存
            statement_cache_size=0,
        )

    async def batch_insert(self, table: str, records: list[dict], batch_size: int = 500):
        """分批批量插入，避免大事务"""
        async with self.pool.acquire() as conn:
            for i in range(0, len(records), batch_size):
                batch = records[i:i + batch_size]
                columns = batch[0].keys()
                values = [
                    tuple(record[col] for col in columns)
                    for record in batch
                ]
                placeholders = ",".join(
                    f"({','.join(f'${j+1}' for j in range(len(columns)))})"
                    for _ in batch
                )
                query = f"""
                    INSERT INTO {table} ({','.join(columns)})
                    VALUES {placeholders}
                    ON CONFLICT (id) DO UPDATE SET
                    {','.join(f"{col}=EXCLUDED.{col}" for col in columns)}
                """
                # 扁平化参数列表
                params = [v for row in values for v in row]
                await conn.execute(query, *params)

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# 初始化 Tracer
provider = TracerProvider()
processor = BatchSpanProcessor(
    OTLPSpanExporter(endpoint="http://otel-collector:4317")
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

app = FastAPI(lifespan=lifespan)

# 自动埋点：路由、请求参数、响应状态
FastAPIInstrumentor.instrument_app(app)

# 手动埋点：业务关键路径
@app.get("/orders/{order_id}")
async def get_order(order_id: str):
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span("check_order_cache") as span:
        span.set_attribute("order.id", order_id)
        cached = await redis.get(f"order:{order_id}")
        if cached:
            span.set_attribute("cache.hit", True)
            return json.loads(cached)
        span.set_attribute("cache.hit", False)
    # ... 数据库查询

from prometheus_fastapi_instrumentator import Instrumentator

# 一键接入，自动注册 /metrics 端点
Instrumentator(
    should_group_status_codes=False,
    should_ignore_untemplated=True,
    should_respect_env_var=True,
    env_var_name="ENABLE_METRICS",
).instrument(app).expose(app)

# .github/workflows/deploy.yml
name: FastAPI CI/CD Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_DB: test
          POSTGRES_PASSWORD: test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
      redis:
        image: redis:7-alpine
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"

      - name: Install dependencies
        run: |
          pip install -r requirements-dev.txt
          pip install -r requirements.txt

      - name: Run lint & type check
        run: |
          ruff check app/
          mypy app/

      - name: Run tests with coverage
        run: |
          pytest tests/ \
            --cov=app \
            --cov-report=xml \
            --cov-report=term-missing \
            -v --tb=short

      - name: Upload coverage
        uses: codecov/codecov-action@v4

  build-and-deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v4

      - name: Build Docker image
        run: |
          docker build \
            --build-arg BUILD_TIME=$(date -u +'%Y-%m-%dT%H:%M:%SZ') \
            --build-arg GIT_COMMIT=${{ github.sha }} \
            -t registry.example.com/fastapi-app:${{ github.sha }} \
            -t registry.example.com/fastapi-app:latest .

      - name: Push to registry
        run: |
          docker push registry.example.com/fastapi-app:${{ github.sha }}
          docker push registry.example.com/fastapi-app:latest

      - name: Deploy to k8s
        run: |
          kubectl set image deployment/fastapi-app \
            fastapi-app=registry.example.com/fastapi-app:${{ github.sha }}
          kubectl rollout status deployment/fastapi-app --timeout=5m

- 单元测试（60%）：pytest + httpx 测试独立函数
- 集成测试（30%）：testcontainers 启动真实 PostgreSQL/Redis 容器
- E2E 测试（10%）：docker-compose 启动全栈服务进行 API 测试