GraphQL客户端必须用async/await配合异步HTTP客户端（如httpx.AsyncClient），因同步requests会阻塞事件循环；服务端resolver也需用async def并await真正的协程对象，避免同步阻塞操作。
GraphQL查询为什么必须用 async/await 而不是 requests
因为 GraphQL 服务端（如 FastAPI + Strawberry）在处理 resolve 函数时，若内部有数据库查询、HTTP 调用或文件读写，必须用 async 函数；否则会阻塞整个事件循环。用同步的 requests 发起查询，哪怕服务端支持异步，客户端也会变成“假异步”——它只是开了个线程，没真正释放 event loop。
正确做法是：客户端也用异步 HTTP 客户端，比如 aiohttp 或 httpx.AsyncClient，配合 await 调用。
- Strawberry 自身不提供客户端，只负责服务端 schema 和 resolver 的异步支持
-
httpx.AsyncClient是目前最轻量、API 最接近requests的选择，且原生支持 HTTP/2 和连接池 - 别用
asyncio.run()包裹单次请求——它每次新建事件循环，开销大，也不支持复用连接
用 httpx.AsyncClient 发起异步 GraphQL 查询
核心就三步：创建 client、构造 JSON body、await client.post()。注意 Content-Type 必须是 application/json，且 query 字段不能漏掉。
import httpx
import asyncio
<p>query = """
query GetUser($id: ID!) {
user(id: $id) {
name
email
}
}
"""</p><p>async def fetch_user(user_id: str):
async with httpx.AsyncClient() as client:
resp = await client.post(
"<a href="https://www.php.cn/link/fe2e9cb2ea6cb28769e3c4947d156690">https://www.php.cn/link/fe2e9cb2ea6cb28769e3c4947d156690</a>",
json={"query": query, "variables": {"id": user_id}},
headers={"Content-Type": "application/json"}
)
return resp.json()</p><h1>调用示例</h1><p>result = asyncio.run(fetch_user("123"))
-
json=...参数会自动设Content-Type，但显式写更稳妥，尤其调试时 - 变量必须用
variables键传入 dict，不能拼进 query 字符串里（会破坏 GraphQL 类型校验） - 如果服务端返回
400 Bad Request，大概率是variables类型不匹配或缺失，检查 schema 中字段是否为非空（$id: ID!）
在 Strawberry resolver 中 await 数据源（如数据库）
Strawberry 的 resolver 支持 async def，但它不会自动把同步函数变成异步——你得自己确保被 await 的对象是真正的协程对象（比如 database.fetch_one() 而不是 sqlite3.connect().execute()）。
立即学习“Python免费学习笔记（深入）”；
import strawberry
from databases import Database
<p>database = Database("sqlite+aiosqlite:///app.db")</p><p>@strawberry.type
class User:
id: str
name: str
email: str</p><p>@strawberry.type
class Query:
@strawberry.field
async def user(self, id: str) -> User:
row = await database.fetch_one(
"SELECT id, name, email FROM users WHERE id = :id",
{"id": id}
)
if not row:
raise Exception("User not found")
return User(**row)
- 用
databases库而非sqlalchemy原生 ORM，后者默认是同步的；若要用 SQLAlchemy，请选asyncpg+create_async_engine - resolver 返回
None会导致 GraphQL 返回null，但不会报错；建议显式抛异常，让客户端收到errors字段 - 别在 resolver 里直接调
time.sleep()或其他同步阻塞操作，它会拖垮整个 FastAPI 进程
并发多个 GraphQL 查询时，别手动 await 每一个
如果你要查 5 个用户，写 5 个 await fetch_user(...) 是串行的。应该用 asyncio.gather() 并发发起，否则耗时是 5× 单次延迟。
async def fetch_users_concurrently(ids: list[str]):
async with httpx.AsyncClient() as client:
tasks = [
client.post(
"https://www.php.cn/link/fe2e9cb2ea6cb28769e3c4947d156690",
json={"query": query, "variables": {"id": uid}},
)
for uid in ids
]
responses = await asyncio.gather(*tasks)
return [r.json() for r in responses]
-
asyncio.gather()不是魔法，它只是并发调度；如果服务端限流或 DB 连接池太小，仍可能超时或报503 - 避免在
gather里传入未加async with的 client 实例——client 必须在同一个 async context 内复用 - 错误处理要单独做：
gather(..., return_exceptions=True)，否则一个失败整个 batch 就崩了
真正容易被忽略的是连接复用和错误传播：HTTP 客户端没 close、resolver 抛了未捕获异常、variables 类型和 schema 对不上——这些都不会立刻报错，但会在压测或真实用户场景里突然暴露。
