5.4 API管理平台设计

5.4 API管理平台设计

本节导读

本节系统讲解API管理平台的设计与实现，涵盖多供应商管理、API网关、成本分账、用量配额和权限管理等核心功能，帮助读者构建统一、高效、安全的API管理平台。

学习目标

• 掌握API管理平台的整体架构设计原则
• 学会多供应商管理系统的设计与实现
• 理解API网关的核心功能和配置方法
• 掌握成本分账和用量配额的设计与优化
• 学会权限管理和安全控制的实现方法

核心概念

API管理平台架构

API管理平台是一个企业级的API服务管理系统，需要统一管理多个API供应商，提供统一的网关入口，实现成本控制和权限管理。其核心架构包括：

关键技术组件

1. API网关层：统一入口，负责路由、鉴权、限流
2. 多供应商管理：管理多个API供应商，实现负载均衡
3. 成本分账系统：统计各供应商成本，实现成本分配
4. 配额管理：管理API调用配额和限制
5. 权限控制：基于角色的访问控制
6. 监控系统：实时监控API使用情况和性能
7. 日志系统：记录所有API调用日志

分步实战

步骤1：API网关管理系统实现

构建统一API网关管理系统：

# api_gateway_manager.py
import asyncio
import time
import json
import aiohttp
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
import uuid
import logging
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from enum import Enum

class GatewayStatus(Enum):
    """网关状态"""
    ACTIVE = "active"
    INACTIVE = "inactive"
    MAINTENANCE = "maintenance"
    ERROR = "error"

class RouteStatus(Enum):
    """路由状态"""
    ENABLED = "enabled"
    DISABLED = "disabled"
    ERROR = "error"

@dataclass
class APIProvider:
    """API供应商"""
    id: str
    name: str
    endpoint: str
    api_key: str
    rate_limit: int
    cost_per_request: float
    priority: int
    status: GatewayStatus
    config: Dict[str, Any]
    created_at: float
    updated_at: float
    
    def __post_init__(self):
        self.created_at = time.time()
        self.updated_at = time.time()

@dataclass
class APIRoute:
    """API路由"""
    id: str
    name: str
    path: str
    provider_id: str
    method: str
    rate_limit: int
    timeout: int
    status: RouteStatus
    config: Dict[str, Any]
    created_at: float
    updated_at: float
    
    def __post_init__(self):
        self.created_at = time.time()
        self.updated_at = time.time()

@dataclass
class APIGateway:
    """API网关实例"""
    id: str
    name: str
    host: str
    port: int
    status: GatewayStatus
    routes: List[APIRoute]
    providers: List[APIProvider]
    config: Dict[str, Any]
    created_at: float
    updated_at: float
    
    def __post_init__(self):
        self.created_at = time.time()
        self.updated_at = time.time()

class GatewayManager:
    """网关管理器"""
    
    def __init__(self, config: Dict[str, Any] = None):
        self.config = config or {}
        self.gateways: Dict[str, APIGateway] = {}
        self.providers: Dict[str, APIProvider] = {}
        self.routes: Dict[str, APIRoute] = {}
        
        # 配置日志
        self.logger = logging.getLogger(__name__)
        
        # 初始化默认网关
        self._initialize_default_gateway()
    
    def _initialize_default_gateway(self):
        """初始化默认网关"""
        default_gateway = APIGateway(
            id="default",
            name="Default API Gateway",
            host="localhost",
            port=8000,
            status=GatewayStatus.ACTIVE,
            routes=[],
            providers=[],
            config={
                "rate_limiting": True,
                "authentication": True,
                "logging": True,
                "monitoring": True
            },
            created_at=time.time(),
            updated_at=time.time()
        )
        
        self.gateways[default_gateway.id] = default_gateway
    
    async def create_provider(self, provider_data: Dict[str, Any]) -> APIProvider:
        """创建API供应商"""
        provider = APIProvider(
            id=str(uuid.uuid4()),
            name=provider_data["name"],
            endpoint=provider_data["endpoint"],
            api_key=provider_data["api_key"],
            rate_limit=provider_data.get("rate_limit", 100),
            cost_per_request=provider_data.get("cost_per_request", 0.001),
            priority=provider_data.get("priority", 1),
            status=GatewayStatus.ACTIVE,
            config=provider_data.get("config", {}),
            created_at=time.time(),
            updated_at=time.time()
        )
        
        self.providers[provider.id] = provider
        
        self.logger.info(f"创建API供应商: {provider.name} - {provider.id}")
        return provider
    
    async def create_route(self, route_data: Dict[str, Any]) -> APIRoute:
        """创建API路由"""
        route = APIRoute(
            id=str(uuid.uuid4()),
            name=route_data["name"],
            path=route_data["path"],
            provider_id=route_data["provider_id"],
            method=route_data["method"],
            rate_limit=route_data.get("rate_limit", 100),
            timeout=route_data.get("timeout", 30),
            status=RouteStatus.ENABLED,
            config=route_data.get("config", {}),
            created_at=time.time(),
            updated_at=time.time()
        )
        
        # 验证供应商是否存在
        if route.provider_id not in self.providers:
            raise ValueError(f"供应商不存在: {route.provider_id}")
        
        self.routes[route.id] = route
        
        # 添加到网关
        for gateway in self.gateways.values():
            if gateway.status == GatewayStatus.ACTIVE:
                gateway.routes.append(route)
                break
        
        self.logger.info(f"创建API路由: {route.name} - {route.id}")
        return route
    
    async def update_provider(self, provider_id: str, update_data: Dict[str, Any]) -> APIProvider:
        """更新API供应商"""
        if provider_id not in self.providers:
            raise ValueError(f"供应商不存在: {provider_id}")
        
        provider = self.providers[provider_id]
        
        # 更新字段
        if "name" in update_data:
            provider.name = update_data["name"]
        if "endpoint" in update_data:
            provider.endpoint = update_data["endpoint"]
        if "api_key" in update_data:
            provider.api_key = update_data["api_key"]
        if "rate_limit" in update_data:
            provider.rate_limit = update_data["rate_limit"]
        if "cost_per_request" in update_data:
            provider.cost_per_request = update_data["cost_per_request"]
        if "priority" in update_data:
            provider.priority = update_data["priority"]
        if "status" in update_data:
            provider.status = GatewayStatus(update_data["status"])
        if "config" in update_data:
            provider.config.update(update_data["config"])
        
        provider.updated_at = time.time()
        
        self.logger.info(f"更新API供应商: {provider.name} - {provider_id}")
        return provider
    
    async def update_route(self, route_id: str, update_data: Dict[str, Any]) -> APIRoute:
        """更新API路由"""
        if route_id not in self.routes:
            raise ValueError(f"路由不存在: {route_id}")
        
        route = self.routes[route_id]
        
        # 更新字段
        if "name" in update_data:
            route.name = update_data["name"]
        if "path" in update_data:
            route.path = update_data["path"]
        if "provider_id" in update_data:
            # 验证供应商是否存在
            if update_data["provider_id"] not in self.providers:
                raise ValueError(f"供应商不存在: {update_data['provider_id']}")
            route.provider_id = update_data["provider_id"]
        if "method" in update_data:
            route.method = update_data["method"]
        if "rate_limit" in update_data:
            route.rate_limit = update_data["rate_limit"]
        if "timeout" in update_data:
            route.timeout = update_data["timeout"]
        if "status" in update_data:
            route.status = RouteStatus(update_data["status"])
        if "config" in update_data:
            route.config.update(update_data["config"])
        
        route.updated_at = time.time()
        
        self.logger.info(f"更新API路由: {route.name} - {route_id}")
        return route
    
    async def delete_provider(self, provider_id: str) -> bool:
        """删除API供应商"""
        if provider_id not in self.providers:
            raise ValueError(f"供应商不存在: {provider_id}")
        
        # 检查是否有路由使用该供应商
        provider_routes = [
            route for route in self.routes.values()
            if route.provider_id == provider_id
        ]
        
        if provider_routes:
            raise ValueError(f"无法删除供应商，还有{len(provider_routes)}个路由在使用它")
        
        # 删除供应商
        del self.providers[provider_id]
        
        self.logger.info(f"删除API供应商: {provider_id}")
        return True
    
    async def delete_route(self, route_id: str) -> bool:
        """删除API路由"""
        if route_id not in self.routes:
            raise ValueError(f"路由不存在: {route_id}")
        
        # 从网关中移除
        for gateway in self.gateways.values():
            gateway.routes = [route for route in gateway.routes if route.id != route_id]
        
        # 删除路由
        del self.routes[route_id]
        
        self.logger.info(f"删除API路由: {route_id}")
        return True
    
    async def get_provider(self, provider_id: str) -> Optional[APIProvider]:
        """获取API供应商"""
        return self.providers.get(provider_id)
    
    async def get_route(self, route_id: str) -> Optional[APIRoute]:
        """获取API路由"""
        return self.routes.get(route_id)
    
    async def list_providers(self, status: Optional[GatewayStatus] = None) -> List[APIProvider]:
        """列出API供应商"""
        providers = list(self.providers.values())
        
        if status:
            providers = [p for p in providers if p.status == status]
        
        # 按优先级排序
        providers.sort(key=lambda p: p.priority)
        
        return providers
    
    async def list_routes(self, provider_id: Optional[str] = None, 
                         status: Optional[RouteStatus] = None) -> List[APIRoute]:
        """列出API路由"""
        routes = list(self.routes.values())
        
        if provider_id:
            routes = [r for r in routes if r.provider_id == provider_id]
        if status:
            routes = [r for r in routes if r.status == status]
        
        # 按创建时间排序
        routes.sort(key=lambda r: r.created_at, reverse=True)
        
        return routes
    
    async def test_route(self, route_id: str, test_data: Dict[str, Any]) -> Dict[str, Any]:
        """测试API路由"""
        if route_id not in self.routes:
            raise ValueError(f"路由不存在: {route_id}")
        
        route = self.routes[route_id]
        provider = self.providers.get(route.provider_id)
        
        if not provider:
            raise ValueError(f"供应商不存在: {route.provider_id}")
        
        # 构建请求
        headers = {
            "Authorization": f"Bearer {provider.api_key}",
            "Content-Type": "application/json"
        }
        
        url = f"{provider.endpoint}{route.path}"
        
        # 发送测试请求
        try:
            async with aiohttp.ClientSession() as session:
                if route.method.upper() == "GET":
                    async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=route.timeout)) as response:
                        result = await response.text()
                        status_code = response.status
                elif route.method.upper() == "POST":
                    async with session.post(url, headers=headers, json=test_data, timeout=aiohttp.ClientTimeout(total=route.timeout)) as response:
                        result = await response.text()
                        status_code = response.status
                else:
                    raise ValueError(f"不支持的方法: {route.method}")
                
                # 更新路由统计
                route.config["last_test_status"] = status_code
                route.config["last_test_result"] = result[:200]  # 保存前200字符
                route.config["last_test_time"] = time.time()
                
                return {
                    "status": "success",
                    "route_id": route_id,
                    "status_code": status_code,
                    "result": result,
                    "response_time": time.time() - route.config.get("last_test_time", 0)
                }
                
        except Exception as e:
            # 更新路由统计
            route.config["last_test_error"] = str(e)
            route.config["last_test_time"] = time.time()
            
            return {
                "status": "error",
                "route_id": route_id,
                "error": str(e),
                "error_type": type(e).__name__
            }
    
    async def get_gateway_statistics(self) -> Dict[str, Any]:
        """获取网关统计信息"""
        total_providers = len(self.providers)
        active_providers = len([p for p in self.providers.values() if p.status == GatewayStatus.ACTIVE])
        total_routes = len(self.routes)
        active_routes = len([r for r in self.routes.values() if r.status == RouteStatus.ENABLED])
        
        provider_statistics = {}
        for provider in self.providers.values():
            provider_statistics[provider.id] = {
                "name": provider.name,
                "status": provider.status.value,
                "rate_limit": provider.rate_limit,
                "cost_per_request": provider.cost_per_request,
                "priority": provider.priority,
                "route_count": len([r for r in self.routes.values() if r.provider_id == provider.id])
            }
        
        route_statistics = {}
        for route in self.routes.values():
            route_statistics[route.id] = {
                "name": route.name,
                "path": route.path,
                "method": route.method,
                "status": route.status.value,
                "rate_limit": route.rate_limit,
                "timeout": route.timeout,
                "provider_name": self.providers.get(route.provider_id, {}).get("name", "Unknown")
            }
        
        return {
            "total_providers": total_providers,
            "active_providers": active_providers,
            "total_routes": total_routes,
            "active_routes": active_routes,
            "provider_statistics": provider_statistics,
            "route_statistics": route_statistics
        }

# 使用示例
async def gateway_manager_example():
    """网关管理器使用示例"""
    
    # 创建网关管理器
    manager = GatewayManager()
    
    # 创建API供应商
    provider1_data = {
        "name": "OpenAI GPT-4",
        "endpoint": "https://api.openai.com/v1",
        "api_key": "sk-test-key-1",
        "rate_limit": 100,
        "cost_per_request": 0.002,
        "priority": 1,
        "config": {
            "model": "gpt-4",
            "temperature": 0.7
        }
    }
    
    provider2_data = {
        "name": "Anthropic Claude",
        "endpoint": "https://api.anthropic.com",
        "api_key": "sk-test-key-2",
        "rate_limit": 50,
        "cost_per_request": 0.003,
        "priority": 2,
        "config": {
            "model": "claude-3-opus",
            "max_tokens": 4000
        }
    }
    
    provider1 = await manager.create_provider(provider1_data)
    provider2 = await manager.create_provider(provider2_data)
    
    print(f"创建供应商1: {provider1.name} - {provider1.id}")
    print(f"创建供应商2: {provider2.name} - {provider2.id}")
​