Python逆向重构：REST与GraphQL API文档树自动化生成

读API文档读到怀疑人生的场景估计不少人都经历过。我前段时间接到一个任务把公司里一套老REST系统和一套新GraphQL网关的接口汇总成一份结构化的参考文档方便测试和前端同学统一检索。打开文档站点一看几十个模块、上千个Endpoint网页一个个点开复制根本不现实。后来我换了个思路——能不能写个Python爬虫直接从OpenAPI描述文件和GraphQL的Introspection机制里把API结构“逆向重构”成一张完整的文档树这个方案最终跑通了而且产出的不只是一份离线文档而是一张可以查询、可以增量更新的“API宇宙地图”。这篇文章就把完整的实操过程拆开讲清楚包括REST侧怎么抓、GraphQL侧怎么抓、循环引用怎么破、并发频率怎么控以及我踩过的几个不写在官方文档里的坑。1. 为什么要“逆向重构”API文档树而不是去爬网页先解释一下我说的“逆向重构”是什么意思。网上很多爬虫教程抓的是API文档站点的HTML页面然后解析DOM把每个Endpoint的标题、URL、参数抠出来。这种方案看起来简单但有两个致命问题第一HTML结构是给浏览器看的字段经常重复、折叠、懒加载解析规则极其脆弱前端改一版页面爬虫就废了第二文档站点的很多描述是经过二次整理的并不等于接口的真实契约字段类型、必填项、枚举值都有可能在展示层丢失。真正的做法是绕开HTML展示层直接找API体系本身的“自描述机制”。REST体系普遍提供OpenAPI/Swagger描述文件本质是一份JSON或YAML里面包含了所有路径、请求参数、响应模型、枚举、嵌套结构GraphQL体系更直接它内建了一套Introspection反射协议客户端可以用标准查询把整个Schema的类型、字段、参数、引用关系全部拉出来。我们要做的其实就是把这两份机器可读的原始描述下载下来解析成树形结构再合并去重最终得到一份统一的、可直接离线浏览的文档树。这套方案还有个额外收益解析结果是结构化数据不是排版后的文字。你可以把它导入SQLite、生成搜索索引、做接口改动对比、甚至自动生成Mock数据。我后面就是把REST和GraphQL两边解析完的数据统一进了一个节点模型最后渲染成了一个本地可搜索的HTML页面。整个过程大概三百行Python不需要任何重型框架核心就是requests加标准库的json和yaml。2. 先摸清两大体系的“自描述底牌”动手之前必须先搞清楚两种API体系各自的元数据长什么样。这一步很多人会跳过结果写出来的解析器面对真实数据时直接崩。所以我建议你先像我一样用一个最简单的脚本把原始文档拉下来肉眼观察一下结构再决定怎么写解析代码。2.1 RESTOpenAPI就是你要的那张“底牌”大多数REST后端框架都会内置OpenAPI描述文件的导出能力。Spring Boot系一般是/v3/api-docsDjango REST Framework可以用/swagger.json或/openapi.yaml还有一些网关会挂在/openapi/v3下面。入口地址可能不同但数据格式是统一的OpenAPI 3.0规范。打开一份OpenAPI 3.0的JSON核心结构就三块paths所有接口路径和HTTP方法的定义每个方法下又有parameters、requestBody、responses。components.schemas所有数据模型的定义相当于接口的“类”清单。$ref到处可见的引用标记格式类似#/components/schemas/User表示“这里直接引用某个模型”。真正解析时最麻烦的就是$ref。很多模型的字段类型会指向另一个模型另一个模型又指向下一个甚至出现A引用B、B引用A的循环。如果只做一层解析文档树就是断的如果不处理循环递归就会栈溢出。所以解析REST文档的核心任务就是把以$ref串联起来的模型网络展平、去环、拼成一棵树。2.2 GraphQLIntrospection是内建的反射机制GraphQL和REST有个本质区别GraphQL的Schema天然就是一份可查询的数据。任何GraphQL服务只要没有显式关闭Introspection都会响应类似下面这样的查询query { __schema { types { name kind fields { name type { kind name ofType { kind name ofType { kind name ofType { name kind } } } } } } } }这段查询返回的JSON里types数组就是所有类型定义字段的type是一个无限嵌套的ofType链用来表达String、[User!]!这类Leaf、List、Non-Null层层包装的关系。之所以说GraphQL比REST更好抓是因为不需要猜入口地址也不依赖框架——只要知道GraphQL的HTTP端点常见的比如/graphql或/api/graphql发一个POST请求body里带上标准查询字符串即可。不过这里也有个坑Introspection返回的数据量可能非常大尤其当Schema有几百个类型和几千个字段时一次性全查下来会有响应体积过大、超时、被网关限流的问题。后面的章节我会讲怎么把一个巨型查询拆成可控的小查询。3. REST文档树的爬取与解析实战这一节直接上代码。我会按真实项目里的推进顺序来讲先确认入口和格式再解析Schema引用最后构造Endpoint树。3.1 第一步下载入口文件并判断格式第一步不是上来就写解析器而是先写一个探测函数确认目标地址返回的是JSON还是YAML以及根节点里的openapi字段值是什么版本。OpenAPI 2.0和3.0的解析逻辑差别很大不检查版本直接解析后面会连环踩雷。import requests import yaml import json from urllib.parse import urljoin def fetch_openapi_document(base_url: str, endpoints: list[str]) - dict: session requests.Session() session.headers.update({ User-Agent: api-doc-analyzer/1.0, Accept: application/json, application/yaml, text/yaml, }) for endpoint in endpoints: url urljoin(base_url, endpoint) resp session.get(url, timeout30) if resp.status_code ! 200: continue try: data resp.json() if openapi in data or swagger in data: return data except json.JSONDecodeError: data yaml.safe_load(resp.text) if isinstance(data, dict) and (openapi in data or swagger in data): return data raise RuntimeError(没有找到可用的OpenAPI描述文件)提示这里的endpoints列表建议按[/v3/api-docs, /swagger.json, /openapi.json, /openapi.yaml]的顺序探测。不同项目暴露的路径差异很大多试几个总没错。返回的字典里如果带swagger: 2.0那就进入2.0分支它的模型定义在definitions而不是components.schemas。我这次项目里碰到的是3.0所以后面的解析以components.schemas为例但代码我会留出兼容2.0的字段映射。3.2 第二步递归展开$ref把散落的Schema拼成完整的引用树拿到原始字典后真正的核心逻辑开始。我的做法分三步先把components.schemas或definitions里所有模型名到模型定义的映射存进一个字典作为“模型注册表”。写一个递归函数resolve_ref输入一个可能带$ref的JSON片段输出一个完全展开、不含引用标记的普通字典。递归过程中用seen集合记录当前展开路径上的模型名遇到循环引用时不再往下展开而是只保留一个指向模型名的占位标记。class OpenAPISchemaResolver: def __init__(self, raw_doc: dict): self.schemas raw_doc.get(components, {}).get(schemas, {}) if not self.schemas: self.schemas raw_doc.get(definitions, {}) self.registry self.schemas def _resolve_ref(self, ref: str, stack: list[str]): # ref 形如 #/components/schemas/User parts [p for p in ref.split(/) if p and p ! #] model_name parts[-1] if model_name in stack: return {$ref_cycle: model_name} definition self.registry.get(model_name) if definition is None: return {$ref_missing: model_name} return self._resolve_node(definition, stack [model_name]) def _resolve_node(self, node, stack: list[str]): if isinstance(node, dict): if $ref in node: return self._resolve_ref(node[$ref], stack) return { key: self._resolve_node(value, stack) for key, value in node.items() } if isinstance(node, list): return [self._resolve_node(item, stack) for item in node] return node这个设计有几个关键决策stack不是全局visited而是“当前展开路径”。为什么要区分因为模型A引用B、B不再引用A时B可以正常展开但如果B又引用A说明存在循环。如果拿全局集合去重同一个模型第二次出现时就直接跳过就会丢失A→B→C→B这种非循环重复引用的细节。用路径栈既能防止死循环又不会误伤普通复用。循环引用时返回{$ref_cycle: model_name}占位这样后续渲染文档树时能明确看到“这个字段的类型是User而User存在自引用”信息不丢失。解析完所有Schema之后就可以把每个模型在树中的“位置”确定下来。我这里用了一个简单算法从每个根模型没有被其他模型引用的模型出发按引用关系遍历把整个结构挂在树形节点下。但如果你不需要严格的“树”而只想得到“模型关系图”那就直接输出这个展开后的字典集合前端渲染时按索引查询即可。3.3 第三步解析paths构造Endpoint与参数节点模型解析只是半成品最终文档树要能回答“有哪些URL、每个URL支持哪些方法、参数是什么、返回什么”。这一层我写了一个build_endpoint_tree遍历paths下每个path项和每个HTTP方法def build_endpoint_tree(openapi_doc: dict, resolver: OpenAPISchemaResolver): endpoints [] paths openapi_doc.get(paths, {}) for path, path_item in paths.items(): if not isinstance(path_item, dict): continue for method in [get, post, put, patch, delete, options, head]: if method not in path_item: continue operation path_item[method] endpoint { path: path, method: method.upper(), operation_id: operation.get(operationId, ), summary: operation.get(summary, ), parameters: [], request_body: None, responses: [], } # 收集路径参数和Query参数 for param in operation.get(parameters, []): schema param.get(schema, {}) endpoint[parameters].append({ name: param.get(name), in: param.get(in), required: param.get(required, False), schema: resolver._resolve_node(schema, []), }) # 收集请求体模型 request_body operation.get(requestBody, {}) content request_body.get(content, {}) if application/json in content: media content[application/json] endpoint[request_body] resolver._resolve_node(media.get(schema, {}), []) # 收集响应模型 responses operation.get(responses, {}) for status, resp in responses.items(): resp_content resp.get(content, {}) response_schema None if application/json in resp_content: response_schema resolver._resolve_node( resp_content[application/json].get(schema, {}), [] ) endpoint[responses].append({ status: status, description: resp.get(description, ), schema: response_schema, }) endpoints.append(endpoint) return endpoints到这里REST部分得到了三个产物schemas展开字典、endpoints列表、以及每个Endpoint携带的请求/响应模型。这个结构已经足够支撑后面统一文档树的渲染了。4. GraphQL类型宇宙的抓取与关系重建GraphQL侧的抓取和解析是完全不同的套路但同样可以做得非常干净。核心就是两条控制Introspection的查询粒度然后递归重建类型引用图。4.1 构造Introspection查询别试图一次拉全宇宙前面提到过一次全量Introspection的风险。实际项目里Schema里的类型动辄几百个每个字段又有类型链、参数、默认值、弃用信息全量响应可能直接到几MB甚至几十MB。很多网关对单个POST body体积和响应体大小都有限制一次性拉全会有超时、413、429之类的问题。我的办法是分两步走第一步只拉类型清单__schema.types里的name和kind不拉字段。这样能拿到一份“类型名册”同时过滤掉以__开头的内建类型。第二步用__type(name: ...)逐个拉取每个类型的详细信息包括字段、参数、类型链、枚举值、接口实现等。第二步虽然请求次数多但每个请求都很小而且可以并发拉取整体速度反而比一次全量更快。具体查询构造如下FULL_TYPE_QUERY query GetType($name: String!) { __type(name: $name) { name kind description fields { name description args { name type { kind name ofType { kind name ofType { kind name ofType { kind name } } } } } type { kind name ofType { kind name ofType { kind name ofType { kind name } } } } } enumValues { name description } inputFields { name type { kind name ofType { kind name ofType { kind name ofType { kind name } } } } } interfaces { name } possibleTypes { name } } }注意type和args下的ofType嵌套。GraphQL的类型链可以是Non-Null - List - Non-Null - Object这种层层包装写三层ofType基本够用万一不够就再多套一层不影响解析——多出来的层级只会多返回几个null而已。4.2 递归重建类型引用图拿到每个类型的JSON之后要把扁平的“类型名 字段 类别名”转换成树形图。这里的树不是文件系统那种父子树而是“类型A的某个字段的类型是BB的某个字段的类型是C”这种引用关系图。def build_graphql_type_graph(types_raw: list[dict]) - dict[str, dict]: graph {} for t in types_raw: name t.get(name) if not name or name.startswith(__): continue graph[name] { kind: t.get(kind), fields: [], input_fields: [], enum_values: [], interfaces: [i.get(name) for i in t.get(interfaces, []) if i], possible_types: [p.get(name) for p in t.get(possibleTypes, []) if p], } for f in t.get(fields, []) or []: graph[name][fields].append({ name: f.get(name), args: [_simplify_arg(a) for a in f.get(args, []) or []], type_ref: _simplify_type_chain(f.get(type)), }) for v in t.get(enumValues, []) or []: graph[name][enum_values].append(v.get(name)) for inp in t.get(inputFields, []) or []: graph[name][input_fields].append({ name: inp.get(name), type_ref: _simplify_type_chain(inp.get(type)), }) return graph_simplify_type_chain的作用是把ofType嵌套链压平成一个易读的字符串比如[User!]!。实现不复杂def _simplify_type_chain(type_dict): if type_dict is None: return null kind type_dict.get(kind) name type_dict.get(name) of_type type_dict.get(ofType) if of_type: inner _simplify_type_chain(of_type) if kind NON_NULL: return f{inner}! if kind LIST: return f[{inner}] return name or unknown_simplify_arg就是类似处理把每个参数的名字、类型链压缩成一个字典。这样最终每个类型的fields里就会包含类似friends: [User!]!这样的引用整张图可以通过类型名互相跳转。4.3 把REST和GraphQL合并成一张“API宇宙地图”两边数据都解析出来之后最后一步是合并。这一步不一定要写多复杂的算法关键是先定义统一的节点模型再按业务名称做关联。我定义的核心数据结构只有两个ResourceNode代表REST的Endpoint包含method、path、params、request_schema、response_schema。TypeNode代表一个数据类型包含name、kind、fields、sourceREST或GraphQL。合并规则我用了三个策略精确匹配两边都叫User、Order这样的类型直接合并成一个TypeNode字段取并集。命名空间映射REST的CreateUserRequest和GraphQL的CreateUserInput通过人工维护一个同义词典来关联词典放在一个独立Python dict里方便后续扩展。无匹配则保留原样REST独有的模型标注source[REST]GraphQL独有的标注source[GraphQL]。一套简单但实用的合并def merge_api_universe(rest_urls: list, graphql_graph: dict, synonym_map: dict): type_registry {} # 先注册GraphQL类型 for name, node in graphql_graph.items(): type_registry[name] { name: name, kind: node[kind], fields: node[fields], source: [GraphQL], } # 遍历REST endpoint把请求/响应里的schema合并进注册表 for ep in rest_urls: for side in [request_body, responses]: schemas ep[side] if side responses else [ep[request_body]] if not schemas: continue for schema_wrapper in schemas: if isinstance(schema_wrapper, dict): schema schema_wrapper.get(schema) if isinstance(schema_wrapper, dict) else schema_wrapper _merge_schema_into_registry(schema, type_registry, synonym_map) return type_registry_merge_schema_into_registry同时负责把REST解析结果里的内联对象提取成命名类型以及处理$ref_cycle占位。合并完成后我将其dump成一个大JSON再交给离线HTML模板渲染。5. 四个比文档本身更值得注意的坑解析代码本身不复杂真正烧时间的是这几个藏在细节里的问题。逐个说清楚帮你避开我踩过的坑。5.1 循环引用导致的死循环无论是OpenAPI的$ref还是GraphQL的类型引用递归解析时都会遇到循环。典型场景一个TreeNode类型的children字段类型是[TreeNode]。如果不加保护解析器会无限递归直到栈溢出。我在OpenAPISchemaResolver里用的stack方案可以防递归死循环但有个副作用第一次遇到循环引用后后续同一路径上再次出现的模型会被折叠成占位标记。渲染时这个占位标记很重要它直接告诉你“这里存在自引用”否则有些前端会在渲染树的时候自己写死递归探测逻辑导致页面卡死。GraphQL侧同理在build_graphql_type_graph阶段我们只是记录了type_ref字符串没有实际递归展开所以不存在循环展开问题——真正递归发生在把图渲染成树视图时这时候一定要在渲染器里维护一个visited集合同一类型在一条展示路径上只展开一次。5.2 请求频率、超时与认证爬取GraphQL类型详情时几百个类型意味着几百个POST请求。如果不做任何控制很快就会被网关限流。我使用的手段是三件套requests.Session复用连接、ThreadPoolExecutor控制并发数量、Semaphore限制全局并发数。from concurrent.futures import ThreadPoolExecutor, as_completed from threading import Semaphore def fetch_all_types(session, endpoint, type_names, max_workers8): sem Semaphore(max_workers) def fetch_one(name): with sem: resp session.post( endpoint, json{query: FULL_TYPE_QUERY, variables: {name: name}}, timeout30, ) if resp.status_code 200: return name, resp.json().get(data, {}).get(__type) return name, None results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [executor.submit(fetch_one, name) for name in type_names] for future in as_completed(futures): name, data future.result() if data: results[name] data return resultsmax_workers8是实测下来比较稳的数值。如果你不确定对方限流策略可以先用max_workers2跑一遍观察响应时间和有没有429再逐步调大。另外很多内部API文档端点不是匿名可访问的需要带Token或Basic Auth在Session的headers里带上Authorization即可还有少数服务要求请求头带Accept: application/json否则返回HTML登录页而不是JSON。5.3 数据变更与增量更新API文档树这种数据最怕的是手动全量重抓。REST系统的OpenAPI文件还好说GraphQL Schema这种如果每次都在解析前重新拉一遍不仅慢还可能因为临时网络故障导致数据丢失。我的方案是加一层本地缓存用JSON文件把抓到的原始数据保存到cache/目录文件名用md5(endpoint 时间戳)生成。每次先读取缓存再请求远端对比ETag或Last-Modified有变化才重新解析没有变化直接用缓存结果。对于GraphQL Schema更简单的做法是把md5的值换成Schema的版本号很多GraphQL服务会在响应头里带X-Schema-Version之类的字段。如果实在拿不到任何缓存头那就退而求其次抓完后对整个解析结果算一次md5存到meta.json里。下次抓取后先对比这个hash如果一样就直接跳过合并和渲染环节能省下不少时间。5.4 离线浏览与搜索索引解析完成后光有一堆JSON和Python字典是不够的最终要给团队用得有一个能离线打开的界面。我选择生成一个零依赖的HTML文件把所有类型和Endpoint渲染成一个可折叠的树形结构顶部放一个搜索框。搜索功能没有上任何前端框架直接用少量的JavaScript做客户端过滤——对几千个节点的文档来说遍历一遍找关键字完全够用。input idsearch placeholder搜索接口名/类型名/字段名... / div idtree/div script // 渲染代码略核心逻辑是遍历 registry模糊匹配 name 与字段名 /script如果你希望有更强大的全文检索比如按字段描述、参数名、枚举值搜索客户端遍历就有些吃力了。这时建议在Python侧把数据导入SQLite建一个简单的FTS5全文索引。这一步在数据落地前做比渲染HTML之后再加要容易得多。我发现这一步带来的效果非常明显以前测试同学查一个字段要在两个文档站之间来回切换现在一个搜索框全搞定。6. 最后分享几点实操体会这套方案我在模拟项目上完整跑过一遍之后最大的体会是抓API文档这件事真正的工作量不在“爬”而在“建模”。爬虫代码只有几十行但模型合并、循环引用处理、并发控制这些细节才是决定一个脚本能不能从“自用玩具”变成“团队工具”的分水岭。另一个很有用的扩展方向是“版本对比”。既然我们已经把两边的Schema都变成了结构化数据那每次抓取后和上一次结果做diff就变得非常简单——谁新增了接口、谁废弃了字段、谁的响应结构变了全都可以自动生成变更日志。我现在做接口变更通知就用这份数据驱动完全不用手工对照文档页面。如果你也在折腾类似的API文档整合建议按这个顺序动手先下载OpenAPI描述文件和GraphQL Introspection原始数据用代码肉眼检查一遍再写解析器优先处理循环引用最后做合并和渲染。不要一开始就追求UI好看先把数据弄干净后面一切都顺。