如何用Python与Prance库构建OpenAPIv3/Swagger API自动代理生成器

1. 定义数据类映射OpenAPI结构 #

用dataclasses定义对应需求的结构类，方便后续填充解析后的数据：

from dataclasses import dataclass, field
from typing import List, Optional, Dict, TypeVar

Method = TypeVar('Method')

@dataclass
class Schema:
    """API返回对象的结构定义"""
    name: str
    properties: Dict[str, str]  # 可扩展为嵌套Schema等复杂类型

@dataclass
class Parameter:
    name: str
    type: str
    location: str  # 对应OpenAPI中的`in`字段，避开Python关键字

@dataclass
class Get:
    parameters: List[Parameter] = field(default_factory=list)
    body: Optional[Dict] = None
    return_schema: Optional[Schema] = None

@dataclass
class Post:
    parameters: List[Parameter] = field(default_factory=list)
    body: Optional[Dict] = None
    return_schema: Optional[Schema] = None

@dataclass
class Endpoint:
    path: str
    description: str
    methods: List[Method] = field(default_factory=list)

2. 解析Prance返回的Spec数据 #

编写工具函数，将Prance解析的嵌套字典转换为上述数据类实例：

2.1 解析参数 #

def parse_parameters(params_list: List[Dict]) -> List[Parameter]:
    parameters = []
    for param in params_list:
        # 提取参数类型（复杂场景需递归解析schema）
        param_type = param.get('schema', {}).get('type', 'string')
        parameters.append(
            Parameter(
                name=param['name'],
                type=param_type,
                location=param['in']
            )
        )
    return parameters

2.2 解析返回Schema #

def parse_response_schema(response: Dict) -> Optional[Schema]:
    # 优先处理200状态码响应，可扩展支持多状态码
    content = response.get('200', {}).get('content', {})
    if not content:
        return None
    # 默认处理application/json类型
    schema = content.get('application/json', {}).get('schema', {})
    if not schema:
        return None
    schema_name = schema.get('title', 'AnonymousSchema')
    # 提取属性及类型，复杂Schema需递归处理
    properties = {k: v.get('type', 'unknown') for k, v in schema.get('properties', {}).items()}
    return Schema(name=schema_name, properties=properties)

2.3 遍历Paths生成Endpoint列表 #

def parse_endpoints(spec: Dict) -> List[Endpoint]:
    endpoints = []
    paths = spec.get('paths', {})
    for path, path_data in paths.items():
        description = path_data.get('description', '')
        methods = []
        # 处理GET方法
        if 'get' in path_data:
            get_data = path_data['get']
            params = parse_parameters(get_data.get('parameters', []))
            return_schema = parse_response_schema(get_data.get('responses', {}))
            methods.append(Get(parameters=params, return_schema=return_schema))
        # 处理POST方法
        if 'post' in path_data:
            post_data = path_data['post']
            params = parse_parameters(post_data.get('parameters', []))
            # 解析请求体
            body_schema = post_data.get('requestBody', {}).get('content', {}).get('application/json', {}).get('schema')
            return_schema = parse_response_schema(post_data.get('responses', {}))
            methods.append(Post(parameters=params, body=body_schema, return_schema=return_schema))
        # 可扩展PUT、DELETE、PATCH等其他HTTP方法
        endpoints.append(Endpoint(path=path, description=description, methods=methods))
    return endpoints

3. 动态生成API代理类 #

基于解析后的Endpoint列表，创建可直接调用的API客户端类：

import requests

def generate_api_client(endpoints: List[Endpoint]) -> type:
    client_methods = {}

    for endpoint in endpoints:
        for method in endpoint.methods:
            # 生成GET方法的代理函数
            if isinstance(method, Get):
                def get_func(self, **kwargs):
                    url = f"{self.base_url}{endpoint.path}"
                    # 替换路径参数（如/{id}）
                    for param in method.parameters:
                        if param.location == 'path':
                            url = url.replace(f"{{{param.name}}}", str(kwargs.pop(param.name)))
                    # 提取查询参数
                    query_params = {p.name: kwargs[p.name] for p in method.parameters if p.location == 'query'}
                    response = self.session.get(url, params=query_params)
                    response.raise_for_status()  # 抛出HTTP错误
                    return response.json()
                # 生成方法名（如get_api_v1_products对应/api/v1/products）
                method_name = f"get_{endpoint.path.strip('/').replace('/', '_')}"
                client_methods[method_name] = get_func

            # 生成POST方法的代理函数
            elif isinstance(method, Post):
                def post_func(self, **kwargs):
                    url = f"{self.base_url}{endpoint.path}"
                    # 替换路径参数
                    for param in method.parameters:
                        if param.location == 'path':
                            url = url.replace(f"{{{param.name}}}", str(kwargs.pop(param.name)))
                    query_params = {p.name: kwargs[p.name] for p in method.parameters if p.location == 'query'}
                    body = kwargs.get('body')
                    response = self.session.post(url, json=body, params=query_params)
                    response.raise_for_status()
                    return response.json()
                method_name = f"post_{endpoint.path.strip('/').replace('/', '_')}"
                client_methods[method_name] = post_func

    # 动态创建APIClient类
    return type('APIClient', (object,), {
        '__init__': lambda self, base_url: (
            setattr(self, 'base_url', base_url),
            setattr(self, 'session', requests.Session())
        ),
        **client_methods
    })

4. 完整使用流程 #

from prance import ResolvingParser

# 1. 解析Swagger文档
parser = ResolvingParser('http://your-swagger-url/swagger.json')
# 2. 生成Endpoint列表
endpoints = parse_endpoints(parser.specification)
# 3. 生成API客户端
APIClient = generate_api_client(endpoints)
# 4. 实例化并调用
client = APIClient('http://api.example.com')
# 调用自动生成的方法
products = client.get_api_v1_products()
# 带参数的调用
single_product = client.get_api_v1_products_id(id=123)

优化与扩展建议 #

• 复杂Schema处理：递归解析嵌套Schema、枚举、数组类型，结合pydantic生成数据模型，实现请求/响应的自动校验。
• 类型提示：为动态生成的方法添加类型注解，提升IDE支持。
• 错误处理：扩展异常捕获逻辑，处理网络错误、超时、非2xx响应等场景。
• 认证支持：在Client类中添加API Key、OAuth2等认证方法，比如在__init__中设置请求头。
• 代码生成：用Jinja2模板渲染生成静态Python代码文件，而非动态创建类，方便调试和版本控制。

替代方案 #

• datamodel-code-generator：从OpenAPI Spec生成Pydantic模型，可基于这些模型封装API调用逻辑。
• 手动代码生成：编写Jinja2模板，直接渲染出包含所有API方法的Python文件，灵活性更高。

内容的提问来源于stack exchange，提问作者RobertLD