如何在Python中实现异步的GraphQL查询_使用strawberry等异步支持库-Python教程-PHP中文网

GraphQL客户端必须用async/await配合异步HTTP客户端（如httpx.AsyncClient），因同步requests会阻塞事件循环；服务端resolver也需用async def并await真正的协程对象，避免同步阻塞操作。

如何在python中实现异步的graphql查询_使用strawberry等异步支持库

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"))
</p>

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 Hardener

通过完整的安全审查、代码整洁重构、健壮的错误处理、正确的日志记录以及完整的文档字符串覆盖，强化 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)
</p>

用 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]