第8章：API网关设计¶

from fastapi import FastAPI, Request
import httpx

app = FastAPI()

# 路由规则
routes = {
    "/api/users": "http://user-service",
    "/api/orders": "http://order-service",
    "/api/products": "http://product-service"
}

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(request: Request, path: str):
    # 匹配路由
    for route, target in routes.items():
        # 去掉路由前缀的 / 后与请求路径比较
        route_prefix = route.lstrip("/")
        if path.startswith(route_prefix):
            # 转发请求（去掉前缀后拼接目标地址）
            sub_path = path[len(route_prefix):]
            url = f"{target}{sub_path}"
            async with httpx.AsyncClient() as client:
                response = await client.request(
                    method=request.method,
                    url=url,
                    headers=dict(request.headers),
                    content=await request.body()
                )
                return response.json()

    return {"error": "Route not found"}

from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials

    # 验证token
    user = validate_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")

    return user

@app.get("/api/users/{user_id}")
async def get_user(user_id: int, user: dict = Depends(verify_token)):
    # 调用用户服务
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://user-service/users/{user_id}")
        return response.json()

def validate_token(token: str):
    # 验证JWT token
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.InvalidTokenError:
        return None

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.get("/api/users/{user_id}")
@limiter.limit("100/minute")
async def get_user(request: Request, user_id: int):
    # 调用用户服务
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://user-service/users/{user_id}")
        return response.json()

import logging
import time
from prometheus_client import Counter, Histogram  # Counter计数器：统计元素出现次数

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Prometheus指标
request_count = Counter('gateway_requests_total', 'Total requests', ['method', 'endpoint'])
request_duration = Histogram('gateway_request_duration_seconds', 'Request duration')

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(request: Request, path: str):
    start_time = time.time()

    # 记录请求
    logger.info(f"{request.method} {request.url}")

    try:  # try/except捕获异常
        # 转发请求
        response = await forward_request(request, path)

        # 记录成功
        request_count.labels(method=request.method, endpoint=path).inc()
        return response
    except Exception as e:
        # 记录错误
        logger.error(f"Error: {e}")
        raise
    finally:
        # 记录耗时
        duration = time.time() - start_time
        request_duration.observe(duration)

import httpx

# 配置连接池
client = httpx.AsyncClient(
    limits=httpx.Limits(
        max_connections=1000,
        max_keepalive_connections=100
    ),
    timeout=httpx.Timeout(30.0)
)

from fastapi_cache import FastAPICache, Coder
from fastapi_cache.backends.redis import RedisBackend

# 配置缓存
FastAPICache.init(RedisBackend(redis_client), prefix="fastapi-cache")

@app.get("/api/users/{user_id}")
@cache(expire=60)  # 缓存60秒
async def get_user(user_id: int):
    # 调用用户服务
    response = await client.get(f"http://user-service/users/{user_id}")
    return response.json()

from fastapi import Response

@app.get("/api/users/{user_id}")
async def get_user(user_id: int):  # async def定义异步函数；用await调用
    # 调用用户服务
    response = await client.get(f"http://user-service/users/{user_id}")  # await等待异步操作完成

    # 返回压缩响应
    import gzip
    compressed = gzip.compress(response.content)
    return Response(
        content=compressed,
        media_type="application/json",
        headers={"Content-Encoding": "gzip"}
    )