基于GraphQL Schema动态生成Python API请求函数的方案咨询

需要一种能根据GraphQL Schema中定义的查询（Query）和变更（Mutation）动态生成Python函数的方案：

• 获取Schema后自动创建可导入的Python函数，支持传入必要参数调用
• 抽象请求构建流程，实现与GraphQL API无缝交互
• Schema变更时无需手动更新客户端函数定义

此前尝试过ariadne graphql code generator，但它仅支持基于固定参数的查询生成客户端（示例如下），无法满足动态传参需求：

query MyQuery {
    getCar(plate_number: "14-77-ER") {
      addresses
      brand
      mongoObjId
    }
}

一、推荐工具/库 #

1. ariadne-codegen（优化配置） #

放弃固定参数的查询模板，改用带变量的GraphQL查询配合自定义配置，即可生成支持动态传参的函数：

• 编写带变量的查询文件（例如get_car.graphql）：

  query GetCar($plate_number: String!) {
    getCar(plate_number: $plate_number) {
      addresses
      brand
      mongoObjId
    }
  }
  

• 在pyproject.toml中配置生成规则：

  [tool.ariadne-codegen]
  schema_path = "schema.graphql"
  queries_path = "queries/"
  output_path = "generated_client/"
  async_client = true  # 可选，开启异步支持
  

• 生成的函数会自动接收plate_number参数，内部处理变量注入和请求发送逻辑。

2. strawberry-codegen #

Strawberry的代码生成工具支持从Schema直接生成类型安全的Python客户端，自动将Query/Mutation映射为带参数的函数：

• 安装依赖：pip install strawberry-codegen
• 生成命令：strawberry-codegen schema.graphql --output generated_client.py
• 生成的函数自带类型注解，调用时只需传入对应参数，无需手动拼接查询字符串。

3. 自定义动态生成（基于graphql-core） #

如果需要高度定制，可使用graphql-core解析Schema，手动生成Python函数代码：

• 用graphql-core的parse_schema方法解析Schema文件
• 遍历Query和Mutation字段，自动生成包含参数、请求构建逻辑的函数定义
• 将生成的代码写入文件并导入项目，适合复杂场景的定制需求。

二、自动化流程 #

1. Schema同步：
   • 用gql-cli introspect https://your-graphql-api.com/graphql > schema.graphql从远程API拉取最新Schema，或直接绑定本地Schema文件。
2. 集成代码生成到构建流程：
   • 在pyproject.toml的[build-system]中添加生成命令，或用pre-commit钩子，每次提交代码前自动重新生成客户端。
3. 导入使用：
   • 直接导入生成的generated_client模块，调用函数时传入参数即可，无需关心请求细节：

     from generated_client import get_car
     
     # 同步调用
     car = get_car(plate_number="14-77-ER")
     # 异步调用（若启用async_client）
     async def fetch_car():
         car = await get_car(plate_number="14-77-ER")
     

三、异步特性最佳实践 #

1. 选用异步客户端：
   • 配置工具生成异步函数（如ariadne-codegen的async_client=true），配合gql[asyncio]底层库实现异步请求。
2. 连接复用：
   • 异步客户端使用连接池，避免每次请求创建新连接，提升并发性能：

     from generated_client import AsyncClient
     
     async def main():
         async with AsyncClient(url="https://your-api.com/graphql") as client:
             car1 = await client.get_car("14-77-ER")
             car2 = await client.get_car("15-88-FR")
     

3. 统一错误处理：
   • 在异步函数中捕获GraphQLError和网络异常，统一处理错误场景：

     from gql import GraphQLError
     
     async def safe_fetch_car(plate_number):
         try:
             async with AsyncClient(url="...") as client:
                 return await client.get_car(plate_number)
         except GraphQLError as e:
             print(f"GraphQL Error: {e}")
         except Exception as e:
             print(f"Network Error: {e}")
     

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