如何在FastAPI中将多团队API合并为单一OpenAPI规范？

方法1：自定义OpenAPI合并逻辑 #

主API作为网关，实时或定期拉取子API的OpenAPI规范，合并后对外提供统一的文档。

操作步骤： #

• 子API（如team2的服务）正常部署，保留默认的/openapi.json规范暴露路径
• 在主API中重写OpenAPI生成逻辑，拉取子API的规范并合并路径、组件等内容

示例代码：

from fastapi import FastAPI
import httpx

app = FastAPI(title="主API网关")

# 配置子API信息
SUB_APIS = [
    {"service_url": "http://team2-api:8000", "path_prefix": "/org/team2"}
]

def merge_specs(main_spec, sub_api_list):
    merged = main_spec.copy()
    # 初始化components避免KeyError
    if "components" not in merged:
        merged["components"] = {}
    
    # 合并子API路径与组件
    for sub in sub_api_list:
        prefix = sub["path_prefix"]
        sub_spec = sub["spec"]
        
        # 给子API路径添加前缀后合并
        for path, path_details in sub_spec["paths"].items():
            merged_path = f"{prefix}{path}"
            merged["paths"][merged_path] = path_details
        
        # 合并schemas、securitySchemes等组件
        sub_components = sub_spec.get("components", {})
        for comp_type, comp_items in sub_components.items():
            if comp_type not in merged["components"]:
                merged["components"][comp_type] = {}
            merged["components"][comp_type].update(comp_items)
    
    return merged

@app.get("/openapi.json", include_in_schema=False)
async def custom_openapi():
    # 获取主API自身的规范
    main_spec = app.openapi()
    # 拉取所有子API的规范
    sub_api_data = []
    async with httpx.AsyncClient() as client:
        for api in SUB_APIS:
            resp = await client.get(f"{api['service_url']}/openapi.json")
            resp.raise_for_status()
            sub_api_data.append({
                "path_prefix": api["path_prefix"],
                "spec": resp.json()
            })
    # 合并所有规范
    return merge_specs(main_spec, sub_api_data)

方法2：结合反向代理实现文档+请求转发 #

如果主API需要同时承担请求转发的网关职责，可以在合并文档的基础上，添加请求代理逻辑。

代理示例代码： #

from fastapi import Request, Response
import httpx

@app.api_route("/org/team2/{rest_of_path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def proxy_team2(request: Request, rest_of_path: str):
    async with httpx.AsyncClient() as client:
        target_url = f"http://team2-api:8000/{rest_of_path}"
        # 转发请求头（移除host避免子API识别异常）
        forward_headers = {k: v for k, v in request.headers.items() if k.lower() != "host"}
        # 转发请求
        resp = await client.request(
            method=request.method,
            url=target_url,
            headers=forward_headers,
            content=await request.body(),
            params=dict(request.query_params)
        )
        # 返回子API响应
        return Response(
            content=resp.content,
            status_code=resp.status_code,
            headers=dict(resp.headers)
        )

注意事项 #

• 路径前缀必须唯一，避免不同子API的路径冲突
• 组件（如schemas）如果重名，需要各团队统一命名规则，或在合并逻辑中处理重名覆盖问题
• 可给子API规范拉取添加缓存，减少重复请求对性能的影响

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