🚀 API 管理：企业数字化转型的“数字连接器”（附 Python 源码实战）

一、 什么是 API 管理？不止是“接口文档”

1. API 管理的核心定义

2. 为什么现代企业必须重视 API 管理？

graph LR
    A[企业内部系统] --> B[API 网关]
    C[合作伙伴] --> B
    D[移动应用] --> B
    B --> E[微服务集群]
    
    B --> F[限流熔断]
    B --> G[认证授权]
    B --> H[监控分析]
    B --> I[日志审计]

• 企业平均管理 300+ 个 API

• 60% 的企业因 API 安全问题遭受过损失

• 良好的 API 管理可提升开发效率 40%

二、 API 管理平台八大核心功能详解

功能1：API 网关（Gateway）—— 流量总入口

# api_gateway.py
from flask import Flask, request, jsonify, abort
import requests
import time
from functools import wraps
import jwt
from datetime import datetime, timedelta
from collections import defaultdict
import threading
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class APIGateway:
    """轻量级 API 网关实现"""
    
    def __init__(self):
        self.app = Flask(__name__)
        self.app.config['SECRET_KEY'] = 'your-secret-key'
        
        # 路由表：{路径: {目标服务, 方法, 限流规则}}
        self.routes = {
            '/api/v1/users': {
                'target': 'http://user-service:5000',
                'methods': ['GET', 'POST'],
                'rate_limit': {'requests': 100, 'per_seconds': 60}
            },
            '/api/v1/orders': {
                'target': 'http://order-service:6000',
                'methods': ['GET', 'POST', 'PUT'],
                'rate_limit': {'requests': 200, 'per_seconds': 60}
            }
        }
        
        # 限流计数器
        self.rate_limiter = RateLimiter()
        
        # 设置路由
        self.setup_routes()
    
    def setup_routes(self):
        """动态设置所有路由"""
        @self.app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH'])
        def gateway_proxy(path):
            """网关代理：处理所有 API 请求"""
            
            # 1. 认证检查
            auth_result = self.authenticate(request)
            if not auth_result['valid']:
                return jsonify({'error': 'Unauthorized'}), 401
            
            # 2. 路由匹配
            full_path = '/' + path
            route_info = self.find_route(full_path, request.method)
            if not route_info:
                return jsonify({'error': 'Route not found'}), 404
            
            # 3. 限流检查
            client_id = auth_result.get('client_id', 'anonymous')
            if not self.rate_limiter.check_limit(client_id, full_path, route_info['rate_limit']):
                return jsonify({'error': 'Rate limit exceeded'}), 429
            
            # 4. 请求转发
            response = self.forward_request(request, route_info)
            
            # 5. 日志记录
            self.log_request(request, response, auth_result)
            
            return response
        
        # 健康检查端点
        @self.app.route('/health')
        def health_check():
            return jsonify({'status': 'healthy', 'timestamp': datetime.now().isoformat()})
    
    def authenticate(self, request):
        """JWT 认证验证"""
        auth_header = request.headers.get('Authorization')
        
        if not auth_header or not auth_header.startswith('Bearer '):
            return {'valid': False, 'reason': 'No token'}
        
        token = auth_header.split(' ')[1]
        
        try:
            # 验证 JWT
            payload = jwt.decode(
                token, 
                self.app.config['SECRET_KEY'], 
                algorithms=['HS256']
            )
            
            return {
                'valid': True,
                'client_id': payload.get('client_id'),
                'user_id': payload.get('user_id'),
                'scopes': payload.get('scopes', [])
            }
        except jwt.ExpiredSignatureError:
            return {'valid': False, 'reason': 'Token expired'}
        except jwt.InvalidTokenError:
            return {'valid': False, 'reason': 'Invalid token'}
    
    def find_route(self, path, method):
        """查找匹配的路由（支持路径参数）"""
        for route_pattern, route_info in self.routes.items():
            # 简单路径匹配（实际应用应使用正则或前缀树）
            if path.startswith(route_pattern) and method in route_info['methods']:
                return route_info
        return None
    
    def forward_request(self, request, route_info):
        """转发请求到后端服务"""
        target_url = f"{route_info['target']}{request.path}"
        
        # 转发请求头（移除敏感头）
        headers = dict(request.headers)
        headers_to_remove = ['Host', 'Authorization', 'Content-Length']
        for h in headers_to_remove:
            headers.pop(h, None)
        
        # 添加追踪头
        headers['X-Request-ID'] = request.headers.get('X-Request-ID', 'gateway-' + str(int(time.time())))
        headers['X-Forwarded-For'] = request.remote_addr
        
        # 转发请求
        resp = requests.request(
            method=request.method,
            url=target_url,
            headers=headers,
            data=request.get_data(),
            params=request.args,
            cookies=request.cookies,
            timeout=30
        )
        
        # 构建响应
        response = jsonify(resp.json()) if resp.headers.get('Content-Type') == 'application/json' else resp.content
        response.status_code = resp.status_code
        
        # 转发响应头
        for key, value in resp.headers.items():
            if key.lower() not in ['content-encoding', 'content-length', 'transfer-encoding']:
                response.headers[key] = value
        
        return response
    
    def log_request(self, request, response, auth_result):
        """记录请求日志（可接入ELK等日志系统）"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'method': request.method,
            'path': request.path,
            'client_ip': request.remote_addr,
            'client_id': auth_result.get('client_id'),
            'user_agent': request.headers.get('User-Agent'),
            'status_code': response.status_code,
            'response_time_ms': 0,  # 实际应计算
            'api_key': request.headers.get('X-API-Key')
        }
        
        # 这里可以输出到文件、数据库或消息队列
        print(f"[API Gateway] {log_entry}")
    
    def run(self, host='0.0.0.0', port=8000):
        """启动网关"""
        print(f"🚀 API Gateway 启动于 http://{host}:{port}")
        self.app.run(host=host, port=port, debug=False)

class RateLimiter:
    """分布式限流器（简化版）"""
    
    def __init__(self):
        self.requests = defaultdict(list)  # client_id+path -> [timestamps]
        self.lock = threading.Lock()
    
    def check_limit(self, client_id, path, limit_config):
        """检查是否超过速率限制"""
        key = f"{client_id}:{path}"
        now = time.time()
        
        with self.lock:
            # 清理过期记录
            window_start = now - limit_config['per_seconds']
            self.requests[key] = [t for t in self.requests[key] if t > window_start]
            
            # 检查是否超限
            if len(self.requests[key]) >= limit_config['requests']:
                return False
            
            # 记录本次请求
            self.requests[key].append(now)
            return True

# 使用示例
if __name__ == "__main__":
    gateway = APIGateway()
    gateway.run(port=8000)

功能2：开发者门户（Developer Portal）

# developer_portal.py
from flask import Flask, render_template, jsonify, request
import yaml
import json
from datetime import datetime

class DeveloperPortal:
    """开发者门户：API 文档、测试、注册"""
    
    def __init__(self):
        self.app = Flask(__name__, template_folder='templates')
        self.apis = self.load_api_specs()
        self.setup_routes()
    
    def load_api_specs(self):
        """加载 API 规格（支持 OpenAPI/Swagger）"""
        return {
            'user-service': {
                'name': '用户服务',
                'version': 'v1.0',
                'description': '用户注册、登录、管理',
                'endpoints': [
                    {
                        'path': '/api/v1/users',
                        'method': 'POST',
                        'summary': '创建用户',
                        'parameters': [
                            {'name': 'username', 'type': 'string', 'required': True},
                            {'name': 'email', 'type': 'string', 'required': True}
                        ],
                        'responses': {
                            '201': {'description': '用户创建成功'},
                            '400': {'description': '参数错误'}
                        }
                    }
                ],
                'rate_limit': '100次/小时',
                'authentication': 'JWT Bearer Token'
            }
        }
    
    def setup_routes(self):
        """设置门户路由"""
        
        @self.app.route('/')
        def index():
            """门户首页"""
            return render_template('index.html', 
                                 apis=self.apis,
                                 total_apis=len(self.apis))
        
        @self.app.route('/api/<api_name>')
        def api_detail(api_name):
            """API 详情页"""
            api_spec = self.apis.get(api_name)
            if not api_spec:
                return "API 不存在", 404
            return render_template('api_detail.html', api=api_spec)
        
        @self.app.route('/api/<api_name>/try', methods=['POST'])
        def try_api(api_name):
            """在线测试 API"""
            api_spec = self.apis.get(api_name)
            if not api_spec:
                return jsonify({'error': 'API not found'}), 404
            
            # 获取测试参数
            data = request.json
            endpoint_path = data.get('endpoint')
            method = data.get('method', 'GET')
            params = data.get('params', {})
            
            # 这里实现实际的 API 调用测试
            result = self.make_test_request(api_name, endpoint_path, method, params)
            
            return jsonify(result)
        
        @self.app.route('/api/keys/register', methods=['POST'])
        def register_api_key():
            """注册 API Key"""
            data = request.json
            client_name = data.get('client_name')
            email = data.get('email')
            
            if not client_name or not email:
                return jsonify({'error': '参数不完整'}), 400
            
            # 生成 API Key
            api_key = self.generate_api_key()
            
            # 保存到数据库（这里简化）
            key_info = {
                'api_key': api_key,
                'client_name': client_name,
                'email': email,
                'created_at': datetime.now().isoformat(),
                'rate_limit': '1000/天',
                'status': 'active'
            }
            
            return jsonify({
                'success': True,
                'api_key': api_key,
                'message': '请妥善保存此 API Key'
            })
        
        @self.app.route('/docs/openapi/<api_name>.json')
        def openapi_spec(api_name):
            """生成 OpenAPI 规范"""
            api_spec = self.apis.get(api_name)
            if not api_spec:
                return jsonify({'error': 'Not found'}), 404
            
            # 转换为 OpenAPI 格式
            openapi_spec = {
                'openapi': '3.0.0',
                'info': {
                    'title': api_spec['name'],
                    'version': api_spec['version'],
                    'description': api_spec['description']
                },
                'paths': self.generate_paths(api_spec)
            }
            
            return jsonify(openapi_spec)
    
    def make_test_request(self, api_name, endpoint, method, params):
        """模拟 API 测试请求"""
        # 实际应该真正调用 API
        return {
            'success': True,
            'endpoint': endpoint,
            'method': method,
            'request_params': params,
            'mock_response': {
                'id': 123,
                'status': 'success',
                'timestamp': datetime.now().isoformat()
            },
            'note': '这是模拟响应，实际会调用真实 API'
        }
    
    def generate_api_key(self):
        """生成 API Key"""
        import secrets
        import hashlib
        import time
        
        raw_key = f"{secrets.token_hex(16)}-{int(time.time())}"
        return hashlib.sha256(raw_key.encode()).hexdigest()[:32]
    
    def generate_paths(self, api_spec):
        """生成 OpenAPI paths 部分"""
        paths = {}
        for endpoint in api_spec.get('endpoints', []):
            paths[endpoint['path']] = {
                endpoint['method'].lower(): {
                    'summary': endpoint['summary'],
                    'parameters': [
                        {
                            'name': param['name'],
                            'in': 'query' if endpoint['method'] == 'GET' else 'body',
                            'required': param.get('required', False),
                            'schema': {'type': param['type']}
                        }
                        for param in endpoint.get('parameters', [])
                    ],
                    'responses': endpoint.get('responses', {})
                }
            }
        return paths
    
    def run(self, port=8080):
        """启动开发者门户"""
        print(f"🌐 开发者门户启动于 http://localhost:{port}")
        self.app.run(port=port, debug=True)

# 模板文件 templates/index.html
HTML_TEMPLATE = '''
<!DOCTYPE html>
<html>
<head>
    <title>API 开发者门户</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 40px; }
        .api-card { border: 1px solid #ddd; padding: 20px; margin: 10px 0; border-radius: 8px; }
        .api-name { font-size: 24px; color: #333; }
        .api-desc { color: #666; margin: 10px 0; }
        .badge { background: #4CAF50; color: white; padding: 4px 8px; border-radius: 4px; font-size: 12px; }
    </style>
</head>
<body>
    <h1>API 开发者门户</h1>
    <p>欢迎使用我们的 API 平台，当前共有 {{ total_apis }} 个 API 服务</p>
    
    {% for api_id, api in apis.items() %}
    <div class="api-card">
        <div class="api-name">{{ api.name }} 
            <span class="badge">{{ api.version }}</span>
        </div>
        <div class="api-desc">{{ api.description }}</div>
        <a href="/api/{{ api_id }}">查看详情与文档</a> |
        <a href="/docs/openapi/{{ api_id }}.json">OpenAPI 规范</a>
    </div>
    {% endfor %}
</body>
</html>
'''

# 保存模板
import os
os.makedirs('templates', exist_ok=True)
with open('templates/index.html', 'w', encoding='utf-8') as f:
    f.write(HTML_TEMPLATE)

# 启动门户
if __name__ == "__main__":
    portal = DeveloperPortal()
    portal.run(port=8080)

功能3：API 生命周期管理

# api_lifecycle.py
from enum import Enum
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
import json
from typing import List, Dict, Optional
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class APIStatus(Enum):
    DESIGN = "design"      # 设计中
    DEVELOP = "develop"    # 开发中
    TEST = "test"         # 测试中
    STAGING = "staging"   # 预发布
    PRODUCTION = "production"  # 生产环境
    DEPRECATED = "deprecated"  # 已弃用
    RETIRED = "retired"   # 已退役

class APIVersionPolicy(Enum):
    NONE = "none"         # 无版本控制
    URI_VERSION = "uri"   # 路径版本 /v1/
    HEADER_VERSION = "header"  # 请求头版本
    PARAM_VERSION = "param"   # 查询参数版本

@dataclass
class APIEndpoint:
    """API 端点定义"""
    path: str
    method: str
    summary: str
    description: str = ""
    request_schema: Optional[Dict] = None
    response_schema: Optional[Dict] = None
    deprecated: bool = False

@dataclass
class APIDefinition:
    """API 定义"""
    id: str
    name: str
    version: str
    description: str
    owner: str
    status: APIStatus
    base_path: str
    endpoints: List[APIEndpoint]
    version_policy: APIVersionPolicy = APIVersionPolicy.URI_VERSION
    created_at: datetime = None
    updated_at: datetime = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()
        if self.updated_at is None:
            self.updated_at = datetime.now()
    
    def to_dict(self):
        """转换为字典（可序列化）"""
        data = asdict(self)
        data['status'] = self.status.value
        data['version_policy'] = self.version_policy.value
        data['created_at'] = self.created_at.isoformat()
        data['updated_at'] = self.updated_at.isoformat()
        return data

class APILifecycleManager:
    """API 生命周期管理器"""
    
    def __init__(self):
        self.apis: Dict[str, APIDefinition] = {}
        self.load_from_storage()
    
    def create_api(self, api_def: APIDefinition) -> str:
        """创建新 API"""
        if api_def.id in self.apis:
            raise ValueError(f"API {api_def.id} 已存在")
        
        # 验证状态流转
        if api_def.status != APIStatus.DESIGN:
            raise ValueError("新 API 必须从 DESIGN 状态开始")
        
        self.apis[api_def.id] = api_def
        self.save_to_storage()
        
        print(f"✅ 创建 API: {api_def.name} v{api_def.version}")
        return api_def.id
    
    def update_status(self, api_id: str, new_status: APIStatus, notes: str = ""):
        """更新 API 状态"""
        if api_id not in self.apis:
            raise ValueError(f"API {api_id} 不存在")
        
        api = self.apis[api_id]
        old_status = api.status
        
        # 验证状态流转规则
        if not self.can_transition(old_status, new_status):
            raise ValueError(f"无法从 {old_status.value} 转换到 {new_status.value}")
        
        api.status = new_status
        api.updated_at = datetime.now()
        
        # 记录状态变更日志
        self.log_status_change(api_id, old_status, new_status, notes)
        
        # 状态特定的操作
        if new_status == APIStatus.PRODUCTION:
            self.publish_to_gateway(api)
        elif new_status == APIStatus.DEPRECATED:
            self.notify_deprecation(api)
        elif new_status == APIStatus.RETIRED:
            self.remove_from_gateway(api)
        
        self.save_to_storage()
        print(f"🔄 状态更新: {api.name} {old_status.value} -> {new_status.value}")
    
    def can_transition(self, from_status: APIStatus, to_status: APIStatus) -> bool:
        """检查状态流转是否允许"""
        transitions = {
            APIStatus.DESIGN: [APIStatus.DEVELOP, APIStatus.RETIRED],
            APIStatus.DEVELOP: [APIStatus.TEST, APIStatus.DESIGN],
            APIStatus.TEST: [APIStatus.STAGING, APIStatus.DEVELOP],
            APIStatus.STAGING: [APIStatus.PRODUCTION, APIStatus.TEST],
            APIStatus.PRODUCTION: [APIStatus.DEPRECATED, APIStatus.STAGING],
            APIStatus.DEPRECATED: [APIStatus.RETIRED, APIStatus.PRODUCTION],
            APIStatus.RETIRED: []  # 退役后不可再更改
        }
        return to_status in transitions.get(from_status, [])
    
    def create_new_version(self, api_id: str, new_version: str) -> str:
        """创建 API 新版本"""
        if api_id not in self.apis:
            raise ValueError(f"API {api_id} 不存在")
        
        old_api = self.apis[api_id]
        
        # 标记旧版本为已弃用
        self.update_status(api_id, APIStatus.DEPRECATED, 
                          f"被新版本 {new_version} 替代")
        
        # 创建新版本
        new_api_id = f"{api_id.split('_')[0]}_{new_version.replace('.', '_')}"
        new_api = APIDefinition(
            id=new_api_id,
            name=old_api.name,
            version=new_version,
            description=old_api.description + f" (版本 {new_version})",
            owner=old_api.owner,
            status=APIStatus.DESIGN,
            base_path=old_api.base_path,
            endpoints=old_api.endpoints,
            version_policy=old_api.version_policy
        )
        
        self.create_api(new_api)
        
        # 记录版本关系
        self.record_version_relationship(api_id, new_api_id)
        
        return new_api_id
    
    def publish_to_gateway(self, api: APIDefinition):
        """发布到 API 网关"""
        print(f"🚀 发布到网关: {api.name} v{api.version}")
        # 实际实现中，这里会调用网关的 API
        # gateway_api.add_route(api)
    
    def notify_deprecation(self, api: APIDefinition):
        """通知 API 弃用"""
        sunset_date = datetime.now() + timedelta(days=90)  # 90天后完全退役
        print(f"⚠️  API 弃用通知: {api.name}")
        print(f"   请迁移到新版本，将在 {sunset_date.strftime('%Y-%m-%d')} 完全退役")
        
        # 在实际中，这里会：
        # 1. 发送邮件通知
        # 2. 在开发者门户标记
        # 3. 返回弃用警告头
        # 4. 记录调用统计
    
    def remove_from_gateway(self, api: APIDefinition):
        """从网关移除"""
        print(f"🗑️  从网关移除: {api.name}")
    
    def log_status_change(self, api_id: str, old_status: APIStatus, 
                         new_status: APIStatus, notes: str):
        """记录状态变更日志"""
        log_entry = {
            'api_id': api_id,
            'timestamp': datetime.now().isoformat(),
            'old_status': old_status.value,
            'new_status': new_status.value,
            'notes': notes
        }
        
        # 写入日志文件或数据库
        with open('api_lifecycle.log', 'a') as f:
            f.write(json.dumps(log_entry) + '\n')
    
    def record_version_relationship(self, old_api_id: str, new_api_id: str):
        """记录版本关系"""
        with open('api_versions.json', 'a') as f:
            relation = {
                'old': old_api_id,
                'new': new_api_id,
                'timestamp': datetime.now().isoformat()
            }
            f.write(json.dumps(relation) + '\n')
    
    def save_to_storage(self):
        """保存到存储"""
        data = {api_id: api.to_dict() for api_id, api in self.apis.items()}
        with open('apis.json', 'w') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
    
    def load_from_storage(self):
        """从存储加载"""
        try:
            with open('apis.json', 'r') as f:
                data = json.load(f)
                
            for api_id, api_data in data.items():
                # 转换状态枚举
                api_data['status'] = APIStatus(api_data['status'])
                api_data['version_policy'] = APIVersionPolicy(api_data.get('version_policy', 'uri'))
                api_data['created_at'] = datetime.fromisoformat(api_data['created_at'])
                api_data['updated_at'] = datetime.fromisoformat(api_data['updated_at'])
                
                # 转换 endpoints
                endpoints = []
                for endpoint_data in api_data.get('endpoints', []):
                    endpoints.append(APIEndpoint(**endpoint_data))
                api_data['endpoints'] = endpoints
                
                self.apis[api_id] = APIDefinition(**api_data)
                
        except FileNotFoundError:
            print("无存储数据，从头开始")
        except Exception as e:
            print(f"加载存储失败: {e}")
    
    def get_api_summary(self):
        """获取 API 概览统计"""
        stats = {
            'total': len(self.apis),
            'by_status': {},
            'by_owner': {}
        }
        
        for api in self.apis.values():
            # 按状态统计
            status = api.status.value
            stats['by_status'][status] = stats['by_status'].get(status, 0) + 1
            
            # 按负责人统计
            owner = api.owner
            stats['by_owner'][owner] = stats['by_owner'].get(owner, 0) + 1
        
        return stats

# 使用示例
if __name__ == "__main__":
    manager = APILifecycleManager()
    
    # 创建示例 API
    user_api = APIDefinition(
        id="user_service_v1_0",
        name="用户服务",
        version="1.0",
        description="用户管理API",
        owner="zhangsan@company.com",
        status=APIStatus.DESIGN,
        base_path="/api/v1",
        endpoints=[
            APIEndpoint(
                path="/users",
                method="POST",
                summary="创建用户",
                description="创建新用户账户"
            ),
            APIEndpoint(
                path="/users/{id}",
                method="GET",
                summary="获取用户信息"
            )
        ]
    )
    
    # 创建 API
    api_id = manager.create_api(user_api)
    
    # 模拟生命周期流转
    print("\n📈 模拟 API 生命周期:")
    manager.update_status(api_id, APIStatus.DEVELOP, "开始开发")
    manager.update_status(api_id, APIStatus.TEST, "进入测试")
    manager.update_status(api_id, APIStatus.STAGING, "预发布环境")
    manager.update_status(api_id, APIStatus.PRODUCTION, "正式上线！")
    
    # 创建新版本
    new_api_id = manager.create_new_version(api_id, "2.0")
    manager.update_status(new_api_id, APIStatus.DEVELOP, "开始开发v2")
    
    # 获取统计
    stats = manager.get_api_summary()
    print(f"\n📊 API 统计: {stats}")

功能4：监控与告警

# api_monitor.py
import time
import threading
import requests
from datetime import datetime, timedelta
from collections import deque
import smtplib
from email.mime.text import MIMEText
from dataclasses import dataclass
from typing import Dict, List, Optional
import json
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
@dataclass
class APIMetric:
    """API 指标"""
    api_name: str
    endpoint: str
    timestamp: datetime
    response_time: float  # 毫秒
    status_code: int
    success: bool
    error_message: Optional[str] = None

class APIMonitor:
    """API 监控与告警系统"""
    
    def __init__(self, check_interval=60):  # 默认60秒检查一次
        self.apis_to_monitor = {}  # API 配置
        self.metrics_history = deque(maxlen=1000)  # 指标历史
        self.alerts_history = []  # 告警历史
        self.check_interval = check_interval
        self.is_running = False
        
        # 告警阈值配置
        self.thresholds = {
            'response_time': 1000.0,  # 1秒
            'error_rate': 0.05,       # 5%
            'availability': 0.99       # 99%
        }
        
        # 告警规则
        self.alert_rules = [
            {
                'name': '高延迟',
                'condition': lambda metrics: any(m.response_time > 5000 for m in metrics[-10:]),  # 最近10次有超过5秒
                'severity': 'warning'
            },
            {
                'name': '高错误率',
                'condition': lambda metrics: self.calculate_error_rate(metrics[-30:]) > 0.1,  # 最近30次错误率>10%
                'severity': 'critical'
            },
            {
                'name': '服务不可用',
                'condition': lambda metrics: not any(m.success for m in metrics[-5:]),  # 最近5次全失败
                'severity': 'critical'
            }
        ]
    
    def add_api_to_monitor(self, api_name: str, config: Dict):
        """添加 API 到监控列表"""
        self.apis_to_monitor[api_name] = config
        print(f"✅ 开始监控 API: {api_name} - {config['url']}")
    
    def start_monitoring(self):
        """启动监控"""
        self.is_running = True
        self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
        self.monitor_thread.start()
        print(f"🚀 API 监控已启动，检查间隔: {self.check_interval}秒")
    
    def stop_monitoring(self):
        """停止监控"""
        self.is_running = False
        if self.monitor_thread:
            self.monitor_thread.join(timeout=5)
        print("🛑 API 监控已停止")
    
    def _monitor_loop(self):
        """监控循环"""
        while self.is_running:
            for api_name, config in self.apis_to_monitor.items():
                try:
                    # 执行健康检查
                    metric = self.check_api_health(api_name, config)
                    self.metrics_history.append(metric)
                    
                    # 检查是否需要告警
                    self.check_alerts(api_name)
                    
                except Exception as e:
                    print(f"❌ 监控 {api_name} 时出错: {e}")
            
            time.sleep(self.check_interval)
    
    def check_api_health(self, api_name: str, config: Dict) -> APIMetric:
        """检查 API 健康状态"""
        url = config['url']
        method = config.get('method', 'GET')
        headers = config.get('headers', {})
        timeout = config.get('timeout', 10)
        
        start_time = time.time()
        try:
            response = requests.request(
                method=method,
                url=url,
                headers=headers,
                timeout=timeout
            )
            response_time = (time.time() - start_time) * 1000  # 转毫秒
            
            success = 200 <= response.status_code < 300
            metric = APIMetric(
                api_name=api_name,
                endpoint=url,
                timestamp=datetime.now(),
                response_time=response_time,
                status_code=response.status_code,
                success=success,
                error_message=None if success else f"HTTP {response.status_code}"
            )
            
            status_icon = "✅" if success else "❌"
            print(f"{status_icon} [{api_name}] {response.status_code} - {response_time:.1f}ms")
            
            return metric
            
        except requests.exceptions.RequestException as e:
            response_time = (time.time() - start_time) * 1000
            metric = APIMetric(
                api_name=api_name,
                endpoint=url,
                timestamp=datetime.now(),
                response_time=response_time,
                status_code=0,
                success=False,
                error_message=str(e)
            )
            print(f"❌ [{api_name}] 请求失败: {e}")
            return metric
    
    def check_alerts(self, api_name: str):
        """检查告警条件"""
        # 获取该 API 的最近指标
        recent_metrics = [
            m for m in self.metrics_history 
            if m.api_name == api_name
        ][-50:]  # 最近50个指标
        
        if len(recent_metrics) < 5:  # 数据太少不检查
            return
        
        for rule in self.alert_rules:
            if rule['condition'](recent_metrics):
                self.trigger_alert(api_name, rule, recent_metrics)
    
    def trigger_alert(self, api_name: str, rule: Dict, metrics: List[APIMetric]):
        """触发告警"""
        alert_id = f"{api_name}_{rule['name']}_{int(time.time())}"
        
        # 检查是否已存在相同告警（防重复）
        recent_alerts = [
            a for a in self.alerts_history[-10:]
            if a.get('api_name') == api_name and a.get('rule_name') == rule['name']
        ]
        
        if recent_alerts and (datetime.now() - recent_alerts[-1]['timestamp']).seconds < 300:
            return  # 5分钟内已有相同告警，抑制
        
        alert = {
            'id': alert_id,
            'timestamp': datetime.now(),
            'api_name': api_name,
            'rule_name': rule['name'],
            'severity': rule['severity'],
            'message': f"[{rule['severity'].upper()}] {api_name} - {rule['name']}",
            'metrics_summary': {
                'total_checks': len(metrics),
                'success_rate': self.calculate_success_rate(metrics),
                'avg_response_time': self.calculate_avg_response_time(metrics),
                'error_rate': self.calculate_error_rate(metrics)
            }
        }
        
        self.alerts_history.append(alert)
        
        # 发送告警
        self.send_alert(alert)
        
        print(f"🚨 告警触发: {alert['message']}")
    
    def send_alert(self, alert: Dict):
        """发送告警通知"""
        # 邮件通知
        if self.should_send_email(alert['severity']):
            self.send_email_alert(alert)
        
        # Webhook 通知
        self.send_webhook_alert(alert)
        
        # 写入日志
        with open('api_alerts.log', 'a') as f:
            f.write(json.dumps(alert, default=str) + '\n')
    
    def send_email_alert(self, alert: Dict):
        """发送邮件告警"""
        try:
            # 配置邮件
            smtp_server = "smtp.company.com"
            smtp_port = 587
            sender = "api-monitor@company.com"
            receivers = ["devops@company.com", "backend-team@company.com"]
            
            # 构建邮件内容
            subject = f"[API告警] {alert['api_name']} - {alert['rule_name']}"
            
            body = f"""
            API 监控告警通知
            =================
            
            告警级别: {alert['severity'].upper()}
            告警时间: {alert['timestamp']}
            API 名称: {alert['api_name']}
            告警规则: {alert['rule_name']}
            
            指标摘要:
            - 检查次数: {alert['metrics_summary']['total_checks']}
            - 成功率: {alert['metrics_summary']['success_rate']:.1%}
            - 平均响应: {alert['metrics_summary']['avg_response_time']:.1f}ms
            - 错误率: {alert['metrics_summary']['error_rate']:.1%}
            
            请及时处理！
            """
            
            msg = MIMEText(body, 'plain', 'utf-8')
            msg['Subject'] = subject
            msg['From'] = sender
            msg['To'] = ', '.join(receivers)
            
            # 发送邮件（实际需要配置SMTP）
            # with smtplib.SMTP(smtp_server, smtp_port) as server:
            #     server.starttls()
            #     server.login(sender, "password")
            #     server.sendmail(sender, receivers, msg.as_string())
            
            print(f"📧 邮件告警已发送: {subject}")
            
        except Exception as e:
            print(f"❌ 发送邮件失败: {e}")
    
    def send_webhook_alert(self, alert: Dict):
        """发送 Webhook 告警（如钉钉、企业微信、Slack）"""
        webhook_url = "https://hooks.example.com/alerts"
        
        # 构建消息
        message = {
            "msgtype": "markdown",
            "markdown": {
                "title": f"API告警 - {alert['severity'].upper()}",
                "text": f"""
                ## 🔥 API 监控告警
                
                **API名称**: {alert['api_name']}
                **告警级别**: {alert['severity'].upper()}
                **告警规则**: {alert['rule_name']}
                **告警时间**: {alert['timestamp']}
                
                **指标摘要**:
                - 成功率: {alert['metrics_summary']['success_rate']:.1%}
                - 平均响应: {alert['metrics_summary']['avg_response_time']:.1f}ms
                
                [点击查看详情](http://monitor.company.com/alerts/{alert['id']})
                """
            }
        }
        
        try:
            # requests.post(webhook_url, json=message, timeout=5)
            print(f"🔔 Webhook 告警已发送: {alert['id']}")
        except:
            print("❌ Webhook 发送失败")
    
    def should_send_email(self, severity: str) -> bool:
        """判断是否需要发送邮件"""
        # 只对 critical 告警发邮件
        return severity == 'critical'
    
    def calculate_success_rate(self, metrics: List[APIMetric]) -> float:
        """计算成功率"""
        if not metrics:
            return 0.0
        successes = sum(1 for m in metrics if m.success)
        return successes / len(metrics)
    
    def calculate_error_rate(self, metrics: List[APIMetric]) -> float:
        """计算错误率"""
        return 1 - self.calculate_success_rate(metrics)
    
    def calculate_avg_response_time(self, metrics: List[APIMetric]) -> float:
        """计算平均响应时间"""
        if not metrics:
            return 0.0
        valid_metrics = [m for m in metrics if m.success]
        if not valid_metrics:
            return 0.0
        return sum(m.response_time for m in valid_metrics) / len(valid_metrics)
    
    def get_metrics_summary(self, hours: int = 1) -> Dict:
        """获取指标摘要"""
        cutoff = datetime.now() - timedelta(hours=hours)
        recent_metrics = [m for m in self.metrics_history if m.timestamp > cutoff]
        
        summary = {}
        for api_name in self.apis_to_monitor.keys():
            api_metrics = [m for m in recent_metrics if m.api_name == api_name]
            if api_metrics:
                summary[api_name] = {
                    'total_checks': len(api_metrics),
                    'success_rate': self.calculate_success_rate(api_metrics),
                    'avg_response_time': self.calculate_avg_response_time(api_metrics),
                    'availability': self.calculate_availability(api_metrics)
                }
        
        return summary
    
    def calculate_availability(self, metrics: List[APIMetric]) -> float:
        """计算可用性（最近1小时）"""
        if not metrics:
            return 0.0
        
        # 模拟每分钟检查一次
        expected_checks = 60
        actual_checks = len(metrics)
        
        if actual_checks < expected_checks * 0.5:  # 监控可能有问题
            return 0.0
        
        success_rate = self.calculate_success_rate(metrics)
        completeness = actual_checks / expected_checks
        
        return success_rate * completeness
    
    def generate_report(self) -> str:
        """生成监控报告"""
        summary = self.get_metrics_summary(hours=24)
        
        report = []
        report.append("=" * 50)
        report.append("📊 API 监控日报")
        report.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append("=" * 50)
        
        for api_name, stats in summary.items():
            report.append(f"\n🔹 {api_name}")
            report.append(f"   检查次数: {stats['total_checks']}")
            report.append(f"   可用性: {stats['availability']:.2%}")
            report.append(f"   平均响应: {stats['avg_response_time']:.1f}ms")
        
        # 告警统计
        recent_alerts = [a for a in self.alerts_history 
                        if (datetime.now() - a['timestamp']).days < 1]
        if recent_alerts:
            report.append(f"\n🚨 24小时告警: {len(recent_alerts)} 个")
            for alert in recent_alerts[-3:]:  # 最近3个
                report.append(f"   - {alert['timestamp'].strftime('%H:%M')} {alert['rule_name']}")
        
        return '\n'.join(report)

# 使用示例
if __name__ == "__main__":
    monitor = APIMonitor(check_interval=10)  # 10秒检查一次
    
    # 添加要监控的 API
    monitor.add_api_to_monitor("用户服务", {
        'url': 'http://localhost:5000/api/v1/users',
        'method': 'GET',
        'headers': {'Authorization': 'Bearer test-token'},
        'timeout': 5
    })
    
    monitor.add_api_to_monitor("订单服务", {
        'url': 'http://localhost:6000/api/v1/orders',
        'method': 'GET',
        'timeout': 5
    })
    
    # 启动监控
    monitor.start_monitoring()
    
    print("监控运行中，按 Ctrl+C 停止...")
    
    try:
        # 模拟运行一段时间
        time.sleep(30)
        
        # 生成报告
        report = monitor.generate_report()
        print("\n" + report)
        
        # 保持运行
        while True:
            time.sleep(1)
            
    except KeyboardInterrupt:
        monitor.stop_monitoring()
        print("\n监控已停止")

功能5-8 的简要实现说明

• 认证：OAuth2.0、JWT、API Key

• 授权：RBAC、ABAC

• 防护：WAF、速率限制、SQL注入检测

• 加密：TLS 1.3、字段级加密

# 使用 Flask + Chart.js
@app.route('/dashboard')
def dashboard():
    metrics = {
        'total_calls': 1000000,
        'success_rate': 99.8,
        'avg_latency': 120,
        'top_apis': ['/users', '/orders', '/products']
    }
    return render_template('dashboard.html', metrics=metrics)

• 基于使用量的计费

• 套餐管理（免费版/专业版/企业版）

• 实时额度检查

• 发票与账单生成

• 语义化版本控制

• 并行多版本支持

• 自动弃用计划

• 迁移指南生成

三、 如何选择 API 管理平台？

四、 最佳实践建议

1. 从 Day 1 开始：不要等有 100 个 API 时才引入管理

2. 标准化先行：制定 API 设计规范（RESTful、错误码、版本）

3. 安全第一：默认启用认证，最小权限原则

4. 监控驱动：建立 SLO（服务等级目标）和告警

5. 文档即代码：OpenAPI 规范纳入代码仓库

6. 渐进式弃用：给客户端足够时间迁移

💡 总结

• 网关是收费站和监控摄像头

• 门户是导航和路标

• 生命周期是道路建设与维护

• 监控是交通状况实时显示

• 安全是交警和交规