Python异步GraphQL实战教程分享

本文深入讲解了如何用 Python 实现高效、健壮的异步 GraphQL 查询，重点推荐 httpx 作为首选 HTTP 客户端——它兼具同步/异步双模式、类 requests 的简洁接口、原生 HTTP/2 支持与自动连接复用，完美适配 Strawberry 的 GraphQL over HTTP 场景；文章手把手演示了正确构造异步请求（必须用 AsyncClient、json= 而非 data=、规范书写 query 字符串、严谨处理 variables 类型）、规避常见陷阱（如未 await response.json()、CSRF/cookie 配置遗漏），并强调服务端无需额外改造即可支持异步调用——前提是 resolver 本身为 async def 且使用异步数据库驱动；最后通过并发查询实战（asyncio.gather + 连接数限制 + 错误穿透处理）揭示：真正的性能瓶颈往往不在请求层，而在于 resolver 中混入同步 I/O、缺乏查询深度/复杂度防护，以及不加节制的批量请求设计——掌握这些细节，才能让异步 GraphQL 从“能跑”真正升级为“高可用、真高效”。

为什么用 httpx 而不是 aiohttp 或 requests

requests 根本不支持异步，硬套 async/await 会直接报 RuntimeWarning: coroutine 'ClientSession.post' was never awaited；aiohttp 虽然能异步，但需要手动管理 ClientSession 生命周期，出错时容易漏关连接；httpx 同时提供同步/异步 API，接口接近 requests，且原生支持 HTTP/2、自动连接复用，和 Strawberry 的 GraphQL over HTTP 场景更匹配。

关键点：

• 必须用 httpx.AsyncClient，不能用 httpx.Client
• GraphQL 请求体必须是 application/json，不能传 data=...（那是表单），得用 json=...
• 如果服务端启用了 CSRF 或 cookie 鉴权，记得设 follow_redirects=True 和 cookies=...

如何构造符合 Strawberry 规范的异步查询请求

Strawberry 默认不加前缀、不强制 schema path，但生产环境常挂载在 /graphql 下。它接受标准 GraphQL POST 请求：JSON body 含 query 字段，可选 variables 和 operationName。

常见错误：

• 把 query 写成字符串但没用三重引号或换行转义 → 解析失败，返回 {"errors": [{"message": "Syntax Error..."}
• 变量类型不匹配（比如后端期待 ID!，却传了 null 或字符串 "123"）→ 400 或空数据
• 忘记 await response.json() → 得到 coroutine object 而非 dict

示例（查询用户）：

import httpx

async def fetch_user(user_id: str):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://localhost:8000/graphql",
            json={
                "query": """
                    query GetUser($id: ID!) {
                        user(id: $id) {
                            name
                            email
                        }
                    }
                """,
                "variables": {"id": user_id},
            },
        )
        return response.json()

Strawberry 服务端要不要额外配置才能支持异步客户端

不需要改服务端代码。Strawberry 的 ASGI app（如用 starlette 或 fastapi 包裹）默认就是异步的，httpx.AsyncClient 发来的请求会被正常路由、解析、执行 resolver —— 前提是你的 resolver 本身是 async def。

注意这几点：

• 如果 resolver 里调了数据库，必须用异步驱动（如 asyncpg、aiosqlite），否则会阻塞事件循环
• 别在 resolver 里写 time.sleep(1) 或 requests.get(...)，这是隐式同步陷阱
• Strawberry 的 @strawberry.field 如果修饰的是普通函数，它会自动在线程池里运行，但性能不如原生 async

如何处理并发多个 GraphQL 查询

直接用 asyncio.gather 并发调用多个 fetch_* 函数即可，比串行快得多。但要注意 Strawberry 服务端是否做了查询限制（如 max_complexity）、连接池是否够用。

实操建议：

• 给 httpx.AsyncClient 加 limits=httpx.Limits(max_connections=100)，避免打开过多 socket
• 如果批量查同一字段（如 100 个 user id），优先考虑用 in 参数或自定义 list resolver，而不是发 100 次单条查询
• 错误处理别只看 response.status_code：GraphQL 即使返回 200，body 里也可能有 "errors" 字段，需主动检查

并发示例：

import asyncio

async def batch_fetch_users(ids: list[str]):
    tasks = [fetch_user(i) for i in ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r for r in results if not isinstance(r, Exception)]

真正麻烦的从来不是发请求，而是 resolver 里混用了同步 I/O、没控制并发数、或者前端传了超长嵌套查询却没配 max_depth。这些地方一松懈，异步就变成假异步。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《Python异步GraphQL实战教程分享》文章吧，也可关注golang学习网公众号了解相关技术文章。