使用python-gql的httpx异步传输执行批量请求时出现AttributeError错误

看起来你遇上的问题是因为AsyncClientSession确实没有execute_batch这个方法——别担心，我来帮你梳理清楚问题根源和解决办法！

问题原因 #

你可能是记错了API方法，或者混淆了旧版本的python-gql用法。在当前的python-gql及gql-transport-httpx版本中，AsyncClientSession并不提供execute_batch这个方法。想要实现异步批量请求，我们需要根据你的GraphQL服务端支持情况，选择两种不同的方案。

解决方案 #

方案1：并发执行独立请求（适用于服务端不支持批量请求）

如果你的GraphQL服务端不接受批量请求（即不能一次发送多个请求对象的数组），我们可以用asyncio.gather来并发执行多个独立的execute调用，这样能达到批量处理的效果：

from gql import Client, GraphQLRequest
from gql.transport.httpx import HTTPXAsyncTransport
from itertools import batched
import asyncio

async def main():
    async with Client(
        transport=HTTPXAsyncTransport(url=GQL_URL, headers=GQL_HEADERS, timeout=3000),
    ) as session:
        # 生成所有需要执行的请求，按1000条数据分批次
        all_requests = [
            GraphQLRequest(document=QUERY, variable_values=batch_vars)
            for batch_vars in batched(DATA, 1000)
        ]
        # 并发执行所有请求
        results = await asyncio.gather(*(session.execute(req) for req in all_requests))
        print(results)

asyncio.run(main())

这种方式相当于同时发起多个独立的GraphQL请求，适合大多数场景，而且不需要服务端做特殊配置。

方案2：使用批量传输（适用于服务端支持批量请求）

如果你的GraphQL服务端明确支持批量请求（可以一次接收一个包含多个请求对象的数组），那我们可以使用专门的批量传输类来优化请求：

from gql import Client, GraphQLRequest
from gql.transport.httpx import HTTPXAsyncBatchTransport  # 替换为批量传输类
from itertools import batched
import asyncio

async def main():
    async with Client(
        transport=HTTPXAsyncBatchTransport(url=GQL_URL, headers=GQL_HEADERS, timeout=3000),
    ) as session:
        # 生成一个批次的请求列表
        batch_requests = [
            GraphQLRequest(document=QUERY, variable_values=batch_vars)
            for batch_vars in batched(DATA, 1000)
        ]
        # 直接传入请求列表给execute方法
        results = await session.execute(batch_requests)
        print(results)

asyncio.run(main())

额外提示 #

别忘记先确认你的依赖包是最新版本，避免因为版本过旧导致功能缺失：

pip install --upgrade gql gql-transport-httpx

内容来源于stack exchange