AI原生开发:ClaudeCode与国产大模型协同实战 第3章:30分钟造个轮子——从0到1构建API网关

# 路由转发配置表：路径前缀 -> 后端服务地址ROUTE_TABLE = {    "/api/users": "https://jsonplaceholder.typicode.com/users",    "/api/posts": "https://jsonplaceholder.typicode.com/posts",}# 限流配置RATE_LIMIT_PER_MINUTE = 60# 鉴权配置AUTH_TOKEN_PREFIX = "Bearer "

from fastapi import Request, HTTPExceptionfrom starlette.middleware.base import BaseHTTPMiddlewarefrom config.settings import AUTH_TOKEN_PREFIXclass AuthMiddleware(BaseHTTPMiddleware):    async def dispatch(self, request: Request, call_next):        # 健康检查路径放行        if request.url.path == "/health":            return await call_next(request)        auth_header = request.headers.get("Authorization")        if not auth_header or not auth_header.startswith(AUTH_TOKEN_PREFIX):            raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")        # 可以在这里解析Token，暂时只校验格式        request.state.user_token = auth_header[len(AUTH_TOKEN_PREFIX):]        response = await call_next(request)        return response

import timefrom collections import defaultdictfrom fastapi import Request, HTTPExceptionfrom starlette.middleware.base import BaseHTTPMiddlewarefrom config.settings import RATE_LIMIT_PER_MINUTEclass RateLimitMiddleware(BaseHTTPMiddleware):    def __init__(self, app):        super().__init__(app)        # 存储IP的访问记录: {ip: [timestamp1, timestamp2, ...]}        self.request_records = defaultdict(list)    async def dispatch(self, request: Request, call_next):        # 健康检查路径放行        if request.url.path == "/health":            return await call_next(request)        client_ip = request.client.host        current_time = time.time()        # 清理1分钟前的记录        self.request_records[client_ip] = [            t for t in self.request_records[client_ip] if current_time - t < 60        ]        # 检查限流        if len(self.request_records[client_ip]) >= RATE_LIMIT_PER_MINUTE:            raise HTTPException(status_code=429, detail="Too many requests")        self.request_records[client_ip].append(current_time)        response = await call_next(request)        return response 

import timeimport loggingfrom fastapi import Requestlogger = logging.getLogger("gateway")class LoggingMiddleware:    def __init__(self, app):        self.app = app    async def __call__(self, request: Request, call_next):        start_time = time.time()        response = await call_next(request)        process_time = (time.time() - start_time) * 1000        formatted_time = f"{process_time:.2f}ms"        logger.info(            f"Method={request.method} Path={request.url.path} "            f"Status={response.status_code} Duration={formatted_time}"        )        return response

import httpxfrom fastapi import APIRouter, Request, HTTPExceptionfrom config.settings import ROUTE_TABLErouter = APIRouter()# 复用 httpx 客户端，性能更好client = httpx.AsyncClient(timeout=10.0)@router.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])async def proxy_request(path: str, request: Request):    # 1. 匹配路由    target_base_url = None    for prefix, backend in ROUTE_TABLE.items():        if f"/{path}".startswith(prefix):            target_base_url = backend            break    if not target_base_url:        raise HTTPException(status_code=404, detail="Route not found")    # 2. 构造目标URL    target_url = f"{target_base_url}/{path.split('/', 2)[-1]}" if "/" in path after first prefix match else target_base_url    # 3. 转发请求    try:        # 复制请求头，去掉 host 和 content-length 避免冲突        headers = dict(request.headers)        headers.pop("host", None)        headers.pop("content-length", None)        body = await request.body()        response = await client.request(            method=request.method,            url=target_url,            headers=headers,            content=body if body else None        )        return {            "code": response.status_code,            "message": "Success",            "data": response.json() if "json" in response.headers.get("content-type", "") else response.text        }    except httpx.RequestError as e:        raise HTTPException(status_code=502, detail=f"Backend service error: {str(e)}")

import loggingimport uvicornfrom fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponsefrom fastapi.middleware.cors import CORSMiddlewarefrom middleware.auth import AuthMiddlewarefrom middleware.ratelimit import RateLimitMiddlewarefrom middleware.logger import LoggingMiddlewarefrom router.proxy import routerfrom config.settings import RATE_LIMIT_PER_MINUTE# 配置日志logging.basicConfig(level=logging.INFO)app = FastAPI(title="Python API Gateway", version="1.0.0")# ========== 全局异常处理 ==========@app.exception_handler(Exception)async def global_exception_handler(request: Request, exc: Exception):    status_code = getattr(exc, "status_code", 500)    if isinstance(exc, Exception) and not hasattr(exc, 'status_code'):        status_code = 500    detail = getattr(exc, "detail", str(exc))    return JSONResponse(        status_code=status_code,        content={"code": status_code, "message": detail, "data": None}    )# ========== 注册中间件（执行顺序：从下到上，即 CORS -> Logging -> RateLimit -> Auth）==========app.add_middleware(AuthMiddleware)app.add_middleware(RateLimitMiddleware)app.add_middleware(LoggingMiddleware)app.add_middleware(    CORSMiddleware,    allow_origins=["*"],    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],)# ========== 注册路由 ==========app.include_router(router)# ========== 健康检查 ==========@app.get("/health")async def health_check():    return {"code": 200, "message": "Gateway is healthy", "data": None}if __name__ == "__main__":    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)