应选用 httpx.AsyncClient，因其同时支持同步/异步、接口类 requests、原生 HTTP/2、自动连接复用，且与 Strawberry 兼容性最佳；需用 AsyncClient、json= 而非 data=、注意 CSRF 和 GraphQL 规范。
直接用 aiohttp + graphql-client 的异步封装是可行的，但容易因手动拼接请求体、忽略连接复用或错误重试逻辑，导致并发下失败率上升或响应延迟突增。
为什么不能直接用 requests.post 调用 GraphQL 异步接口
因为 requests 是同步阻塞的，即使放在 asyncio.to_thread 里跑，也会在高并发时迅速耗尽线程池，且无法利用 HTTP/2 多路复用或连接池复用。真实压测中，QPS 可能比原生异步方案低 3–5 倍，错误日志里常出现 ConnectionResetError 或超时。
- GraphQL 请求本质仍是 POST，但需严格设置
Content-Type: application/json - 必须手动序列化
query、variables到 JSON body，漏掉json.dumps()编码会返回400 Bad Request -
requests.Session不支持 async/await，强行 await 会报RuntimeWarning: coroutine 'Session.post' was never awaited
用 aiohttp.ClientSession 封装 GraphQL 请求的最小可靠写法
核心是复用 session、显式控制超时、统一处理 2xx/4xx 响应，并把 GraphQL 错误（如字段不存在）和网络错误区分开。
import aiohttp
import json
async def gql_post(url: str, query: str, variables: dict = None, timeout: int = 10):
headers = {"Content-Type": "application/json"}
payload = {"query": query}
if variables:
payload["variables"] = variables
timeout_cfg = aiohttp.ClientTimeout(total=timeout)
async with aiohttp.ClientSession(timeout=timeout_cfg) as session:
async with session.post(url, headers=headers, data=json.dumps(payload)) as resp:
if resp.status == 200:
result = await resp.json()
if "errors" in result:
raise RuntimeError(f"GraphQL error: {result['errors']}")
return result["data"]
else:
raise RuntimeError(f"HTTP {resp.status}: {await resp.text()}")
- 不要在每次调用都新建
ClientSession，否则 TCP 连接无法复用，高并发下会触发Too many open files -
json.dumps()必须显式调用，aiohttp不会自动序列化 dict - GraphQL 返回的
errors字段在status == 200时仍可能存在，不能只靠 HTTP 状态码判断成败
并发调用多个 GraphQL 查询时如何避免 ClientConnectorError
默认 aiohttp 连接池大小是 100，但若目标服务端限制了单 IP 并发连接数（如 Nginx 默认 limit_conn），或 DNS 解析慢，就容易在批量请求时触发 ClientConnectorError: Cannot connect to host。
立即学习“Python免费学习笔记（深入）”；
- 显式配置连接池：用
aiohttp.TCPConnector(limit=50, limit_per_host=10)控制并发上限 - 加 DNS 缓存：传入
force_close=False和use_dns_cache=True - 对同一域名的多次查询，优先合并为一个
batch请求（需服务端支持graphql-batch或 Apollo Federation 的@defer） - 避免在循环里无节制
asyncio.create_task()，改用asyncio.gather(*tasks, return_exceptions=True)统一收口
真正难的不是发一次请求，而是当服务端返回部分数据 + extensions: {"delayed": true}，或需要按依赖顺序串行 fetch 若干个 @stream 字段时——这时候光靠封装一个 gql_post 函数远远不够，得结合 graphql-ws 协议或自定义响应解析器。
