当前位置：首页 > news >正文

Python API网关：架构设计与实战

news 2026/5/22 0:52:50

Python API网关：架构设计与实战

引言

在Python开发中，API网关是微服务架构的核心组件。作为一名从Rust转向Python的后端开发者，我深刻体会到API网关在系统集成方面的优势。Python提供了多种工具和框架来构建API网关，包括FastAPI、Flask和专业的API网关解决方案。

API网关核心概念

什么是API网关

API网关是位于客户端和后端服务之间的统一入口点，具有以下特点：

统一入口：所有请求通过网关进入
路由转发：根据路径转发到不同服务
认证授权：统一处理认证和授权
限流熔断：保护后端服务
监控日志：统一的监控和日志

架构设计

┌─────────────────────────────────────────────────────────────┐ │ API网关架构 │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 客户端 │───▶│ API网关 │───▶│ 后端服务 │ │ │ │ (Client) │ │ (Gateway) │ │ (Services) │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ 认证 + 限流 + 监控 + 日志 │ │ │ └──────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘

环境搭建与基础配置

使用FastAPI构建网关

from fastapi import FastAPI, Request, HTTPException import httpx app = FastAPI() @app.api_route("/users/{path:path}", methods=["GET", "POST", "PUT", "DELETE"]) async def proxy_users(request: Request, path: str): async with httpx.AsyncClient() as client: url = f"http://user-service/{path}" response = await client.request( method=request.method, url=url, headers=dict(request.headers), content=await request.body() ) return response.json()

使用Flask构建网关

from flask import Flask, request, jsonify import requests app = Flask(__name__) @app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE']) def proxy(path): url = f"http://backend-service/{path}" response = requests.request( method=request.method, url=url, headers=dict(request.headers), data=request.get_data(), cookies=request.cookies ) return jsonify(response.json()), response.status_code

高级特性实战

认证中间件

from fastapi import FastAPI, Request, HTTPException from fastapi.security import APIKeyHeader app = FastAPI() api_key_header = APIKeyHeader(name="X-API-Key") async def get_api_key(api_key: str = Depends(api_key_header)): if api_key != "secret": raise HTTPException(status_code=401, detail="Invalid API key") return api_key @app.get("/protected") async def protected_route(api_key: str = Depends(get_api_key)): return {"message": "Protected resource"}

限流中间件

from fastapi import FastAPI, Request, HTTPException from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded app = FastAPI() limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) @app.get("/limited") @limiter.limit("5/minute") async def limited_route(request: Request): return {"message": "Limited resource"}

请求日志

from fastapi import FastAPI, Request import time app = FastAPI() @app.middleware("http") async def log_requests(request: Request, call_next): start_time = time.time() response = await call_next(request) duration = time.time() - start_time print(f"{request.method} {request.url.path} - {response.status_code} - {duration:.2f}s") return response

实际业务场景

场景一：多服务路由

from fastapi import FastAPI, Request import httpx app = FastAPI() ROUTES = { "/users": "http://user-service", "/orders": "http://order-service", "/products": "http://product-service", } @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"]) async def proxy(request: Request, path: str): for prefix, service in ROUTES.items(): if path.startswith(prefix.lstrip('/')): url = f"{service}/{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() raise HTTPException(status_code=404, detail="Route not found")

场景二：请求聚合

from fastapi import FastAPI import httpx app = FastAPI() @app.get("/dashboard/{user_id}") async def get_dashboard(user_id: int): async with httpx.AsyncClient() as client: user_task = client.get(f"http://user-service/users/{user_id}") orders_task = client.get(f"http://order-service/orders?user_id={user_id}") products_task = client.get(f"http://product-service/products") user, orders, products = await asyncio.gather( user_task, orders_task, products_task ) return { "user": user.json(), "orders": orders.json(), "products": products.json() }

场景三：请求转换

from fastapi import FastAPI, Request import httpx app = FastAPI() @app.post("/legacy-order") async def create_legacy_order(request: Request): data = await request.json() transformed_data = { "order_id": data.get("id"), "customer": { "name": data.get("customer_name"), "email": data.get("customer_email") }, "items": data.get("products", []) } async with httpx.AsyncClient() as client: response = await client.post( "http://order-service/orders", json=transformed_data ) return response.json()

性能优化

使用连接池

import httpx client = httpx.AsyncClient( limits=httpx.Limits( max_connections=100, max_keepalive_connections=10 ) )

使用缓存

from functools import lru_cache @lru_cache(maxsize=128) async def get_user(user_id: int): async with httpx.AsyncClient() as client: response = await client.get(f"http://user-service/users/{user_id}") return response.json()