REST 的痛点是「一个页面要打七八个接口，每个接口返回一堆用不上的字段」；GraphQL 的解法是「一个端点、客户端自述所需字段」。但把控制权交给客户端，也意味着把 N+1 查询、超深嵌套、查询风暴的风险一并交给了服务端。
Python 生态里写 GraphQL 的主流选择是 Strawberry（基于类型标注、代码优先）与 Graphene（较老、schema 优先）。本文以 Strawberry 为主线，重点不在 API 罗列，而在两个真正决定生产可用性的问题：Schema 怎么设计才不给自己挖坑、N+1 与查询成本怎么控制。
1. GraphQL 与 REST 的取舍
1.1 核心差异
| 维度 | REST | GraphQL |
|---|---|---|
| 端点数量 | 每个资源一个 | 单一端点 |
| 返回字段 | 服务端决定 | 客户端声明 |
| 版本管理 | /v2/ 路径 | 加字段即可，无版本 |
| 缓存 | HTTP 缓存天然可用 | 需持久化查询或 CDN 配合 |
| 错误语义 | HTTP 状态码 | 200 + errors 数组 |
| 过度获取 | 常见 | 基本消除 |
| 请求次数 | 多次往返 | 一次取回 |
| 学习成本 | 低 | 中高 |
GraphQL 解决的是客户端形状多样的问题：同一个后端要服务 Web、iOS、Android、小程序，各端需要的字段差异大。如果只有一个客户端且资源边界清晰，REST 往往更简单。
1.2 什么时候不该用 GraphQL
- 接口纯粹是内部服务间调用（gRPC 更合适）
- 需要 HTTP 缓存与 CDN 边缘缓存（GraphQL 的 POST + 动态查询天然不友好）
- 团队没有 schema 治理能力，容易演变成「一个巨型类型」
- 文件上传、流式响应为主（GraphQL 处理这些很别扭）
2. Strawberry 入门
2.1 安装与最小 schema
uv add strawberry-graphql[fastapi]
import strawberry
from typing import Optional
@strawberry.type
class Author:
id: strawberry.ID
name: str
@strawberry.type
class Book:
id: strawberry.ID
title: str
year: int
author: Author
@strawberry.type
class Query:
@strawberry.field
def book(self, id: strawberry.ID) -> Optional[Book]:
return load_book(id)
@strawberry.field
def books(self, limit: int = 20) -> list[Book]:
return list_books(limit)
schema = strawberry.Schema(query=Query)
类型来自普通 Python 类，@strawberry.type 把字段与标注映射成 GraphQL 类型。strawberry.ID 对应 GraphQL 的 ID 标量（序列化为字符串）。Optional[X] 映射为可空字段，这是 GraphQL 里默认不可空、必须显式标注才可空的语义——与 Python 恰好相反，务必留意。
2.2 Mutation 与 Input 类型
@strawberry.input
class CreateBookInput:
title: str
year: int
author_id: strawberry.ID
@strawberry.type
class Mutation:
@strawberry.mutation
def create_book(self, input: CreateBookInput) -> Book:
return repo.create(input.title, input.year, input.author_id)
schema = strawberry.Schema(query=Query, mutation=Mutation)
用 @strawberry.input 定义入参对象而非直接堆参数，有两个好处：字段可复用、将来加可选字段不破坏已有调用方。命名约定上，单个参数用 input，多个语义独立的参数直接展开。
2.3 与 FastAPI 集成
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
async def get_context() -> dict:
return {"request": None}
graphql_app = GraphQLRouter(schema, context_getter=get_context)
app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")
启动后访问 /graphql 即得到内置的 GraphiQL 交互界面。GraphQLRouter 也支持 graphiql=False 在生产环境关闭 IDE。若你尚未搭好 Web 层，可先参考 Python Web 框架
选型；FastAPI 与 Strawberry 的组合是当前最省心的搭配。
2.4 异步 resolver
@strawberry.type
class Query:
@strawberry.field
async def books(self, limit: int = 20) -> list[Book]:
return await repo.list_async(limit)
Strawberry 完全支持 async def，且与 FastAPI 的事件循环共用。原则是：IO 密集一律 async，CPU 密集交给进程池（否则会阻塞整个事件循环，拖垮并发）。
3. Schema 设计
3.1 可空性是契约
@strawberry.type
class User:
id: strawberry.ID
email: str # 不可空：永远存在
nickname: Optional[str] # 可空：可能未设置
avatar_url: Optional[str] = None
判断标准是「业务上是否可能缺失」，而不是「数据库列是否 NOT NULL」。把「理论上永远有值」的字段标为不可空，客户端就不必写防御代码；反过来把可能缺失的字段标成不可空，一次空值就会让整个查询失败（GraphQL 的可空性错误会冒泡到最近的可空父级）。
3.2 接口与联合
@strawberry.interface
class Node:
id: strawberry.ID
@strawberry.type
class Article(Node):
title: str
@strawberry.type
class Video(Node):
duration: int
@strawberry.type
class Query:
@strawberry.field
def search(self, q: str) -> list[Node]:
...
# 联合类型：成员无共同字段
SearchResult = strawberry.union("SearchResult", (Article, Video))
接口（Interface） 表示「有一组共同字段」；联合（Union） 表示「是其中某一个」。客户端用内联片段（inline fragment）区分具体类型：
query {
search(q: "python") {
__typename
... on Article { title }
... on Video { duration }
}
}
3.3 分页：Relay 连接规范
GraphQL 官方推荐 Relay 的 Connection 模式：
@strawberry.type
class PageInfo:
has_next_page: bool
end_cursor: Optional[str]
@strawberry.type
class BookEdge:
node: Book
cursor: str
@strawberry.type
class BookConnection:
edges: list[BookEdge]
page_info: PageInfo
@strawberry.type
class Query:
@strawberry.field
def books(self, first: int = 20, after: Optional[str] = None) -> BookConnection:
...
| 分页方案 | 优点 | 缺点 |
|---|---|---|
偏移量 offset/limit | 实现简单，可跳页 | 深分页慢，数据变动会错位 |
游标 first/after | 稳定、性能好 | 不能跳页 |
| 全量 + 客户端分页 | 简单 | 数据量大时不可行 |
游标（cursor）通常用「排序键 + 主键」编码，既能保证唯一排序，也能用 WHERE (k, id) > (?, ?) 走索引。这与 Python 数据库与 ORM
中「避免深偏移、改用键集分页」的建议是同一个工程结论。
3.4 枚举与标量
from enum import Enum
import strawberry
import datetime
@strawberry.enum
class Status(Enum):
DRAFT = "draft"
PUBLISHED = "published"
@strawberry.scalar(
serialize=lambda v: v.isoformat(),
parse_value=lambda v: datetime.datetime.fromisoformat(v),
)
class DateTime:
...
自定义标量（scalar）用于精确控制序列化，典型如 DateTime、JSON、Decimal。把 Decimal 直接暴露成 Float 会在金额场景引入浮点误差，必须自定义标量按字符串传输。
3.5 字段命名与描述
@strawberry.type
class User:
created_at: datetime.datetime = strawberry.field(
description="账户创建时间（UTC）",
name="createdAt",
)
约定：GraphQL 字段用 camelCase，Python 用 snake_case，Strawberry 可自动转换；每个公开字段都写 description，它会进入 introspection 结果，是客户端文档的唯一来源。
4. DataLoader 与 N+1
4.1 N+1 是怎么产生的
@strawberry.field
def author(self) -> Author:
return db.get_author(self.author_id) # 每本书查一次作者
查询 100 本书时，GraphQL 会为每本书各调用一次 resolver，于是产生 1（列表）+ 100（作者）= 101 次查询。这是 GraphQL 最常见的性能灾难，且在开发环境数据量小时完全看不出来。
4.2 DataLoader 批处理
from strawberry.dataloader import DataLoader
async def load_authors(keys: list[str]) -> list[Author]:
rows = await db.fetch_authors(keys) # 一次 IN 查询
by_id = {r.id: r for r in rows}
return [by_id.get(k) for k in keys] # 顺序必须与 keys 对应
@strawberry.type
class Book:
author_id: strawberry.Private[str]
@strawberry.field
async def author(self, info: strawberry.Info) -> Author:
return await info.context["author_loader"].load(self.author_id)
async def get_context() -> dict:
return {"author_loader": DataLoader(load_batch=load_authors)}
DataLoader 的核心是在同一事件循环 tick 内收集所有 key，合并成一次批量请求。两个必须遵守的约定：返回列表的顺序必须与输入 keys 严格一致；找不到的 key 要返回 None 占位而不是跳过，否则对应关系会整体错位。
4.3 注意事项
| 坑 | 后果 | 处理 |
|---|---|---|
| 顺序与 keys 不一致 | 数据错位，且不报错 | 用 by_id.get(k) 逐 key 映射 |
| DataLoader 跨请求复用 | 缓存串数据 | 每个请求新建实例 |
| 在同步 resolver 里用 | 无法批处理 | resolver 必须 async |
| 批量 key 过多 | SQL 参数超限 | 分片（如每 500 个一批） |
DataLoader 实例必须请求级创建（放在 context 里），绝不能做成模块级单例，否则会把 A 用户的缓存泄漏给 B 用户。
5. 鉴权、错误与安全
5.1 上下文与字段级权限
from strawberry.permission import BasePermission
class IsAuthenticated(BasePermission):
message = "需要登录"
async def has_permission(self, source, info: strawberry.Info, **kwargs) -> bool:
return info.context["user"] is not None
@strawberry.type
class Query:
@strawberry.field(permission_classes=[IsAuthenticated])
def me(self, info: strawberry.Info) -> User:
return info.context["user"]
权限应当声明在字段上而非塞进 resolver 逻辑，这样既能在 schema 层面审计「哪些字段需要什么权限」，也便于统一测试。对于「只能看自己的数据」这类对象级权限，需要在 resolver 内结合 context 判断。
5.2 错误处理与脱敏
GraphQL 的约定是：HTTP 状态码恒为 200，错误放在 errors 数组里。但绝不能把内部异常原样抛出——数据库错误信息会泄漏表结构。
import strawberry
from strawberry.extensions import SchemaExtension
class ErrorMasking(SchemaExtension):
def on_operation(self):
yield
result = self.execution_context.result
if result and result.errors:
for err in result.errors:
if err.original_error and not isinstance(err.original_error, UserError):
err.message = "内部错误，请稍后重试"
err.extensions.clear()
业务错误（如「库存不足」）应作为数据的一部分返回，而不是塞进 errors：
@strawberry.type
class CreateOrderPayload:
ok: bool
message: Optional[str]
order: Optional[Order]
用 Payload 类型承载业务结果，客户端就能像处理普通字段一样处理失败，不必解析 errors。
5.3 查询成本控制
恶意或粗心的客户端可以用一个查询打垮服务：
query {
users(first: 1000) {
friends(first: 1000) {
friends(first: 1000) { id }
}
}
}
三层嵌套就是十亿级数据。三道防线：
| 手段 | 作用 | 实现 |
|---|---|---|
| 深度限制 | 拒绝过深嵌套 | 解析 AST 计算深度，超过阈值报错 |
| 复杂度/成本分析 | 按字段权重累计成本 | 给列表字段按 first 加权 |
| 分页上限 | 强制 first 有上界 | 服务端裁剪到最大值 |
| 超时 | 兜底 | 请求级超时 |
from strawberry.extensions import QueryDepthLimiter
schema = strawberry.Schema(
query=Query,
extensions=[QueryDepthLimiter(max_depth=8)],
)
# 强制分页上限：resolver 内裁剪
@strawberry.field
def books(self, first: int = 20) -> list[Book]:
return list_books(min(first, 100)) # 客户端传 10000 也只给 100
5.4 生产环境关闭 introspection
graphql_app = GraphQLRouter(
schema,
graphiql=False,
introspection=False, # 生产环境隐藏 schema
)
关闭 introspection 能减少 schema 泄露（攻击者据此构造高成本查询），但会牺牲部分客户端开发体验。折中方案是保留 introspection 但加严格的复杂度限制，或对未认证请求关闭。
5.5 持久化查询
{"id": "a3f1c9", "query": "query Books($n:Int!){books(first:$n){id title}}"}
客户端只发送查询 ID 而非完整查询文本，服务端查表还原。好处是：可以走 GET + CDN 缓存、防止任意查询注入、减小请求体。对公网 API 是强烈推荐的加固手段。
6. 订阅（Subscription）
6.1 定义与推送
import asyncio
from typing import AsyncGenerator
@strawberry.type
class Subscription:
@strawberry.subscription
async def book_added(self) -> AsyncGenerator[Book, None]:
async for book in event_bus.subscribe("book_added"):
yield book
schema = strawberry.Schema(query=Query, mutation=Mutation, subscription=Subscription)
订阅走 WebSocket（graphql-transport-ws 协议）。Strawberry 的 FastAPI 集成内置支持，客户端用 graphql-ws 或 Apollo 的订阅链路连接。
6.2 工程注意事项
| 问题 | 处理 |
|---|---|
| 连接数爆炸 | 限制单用户订阅数、心跳超时踢除 |
| 消息广播风暴 | 按 topic 订阅，只推给关注者 |
| 多实例部署 | 用 Redis Pub/Sub 或 Kafka 做跨实例广播 |
| 鉴权 | 在 WebSocket 握手阶段校验 token |
| 背压 | 有界队列 + 丢弃策略，避免慢客户端拖垮服务 |
单实例内存里的事件总线在多副本部署下会失效——A 实例产生的事件推不到连在 B 实例上的客户端。生产环境必须引入外部消息中间件做广播。
7. 测试与部署
7.1 测试 schema
from strawberry.test import GraphQLTestClient
client = GraphQLTestClient(schema)
def test_books_query():
res = client.query(
"""
query {
books(first: 2) { id title }
}
"""
)
assert not res.errors
assert len(res.data["books"]) == 2
def test_depth_limit():
deep = "query { " + "books { " * 12 + "id" + " }" * 12 + " }"
res = client.query(deep)
assert res.errors
测试要覆盖三类：功能（字段返回正确）、权限（越权访问被拒）、防护（深度/复杂度限制生效）。第三类最容易被漏，却恰恰是生产事故的高发区。
7.2 可观测性
GraphQL 单端点让传统按 URL 统计的监控失效——所有请求都打在 /graphql。因此必须按 operation name 打点：
class MetricsExtension(SchemaExtension):
def on_operation(self):
start = time.perf_counter()
yield
name = self.execution_context.operation_name or "anonymous"
duration = time.perf_counter() - start
metrics.histogram("graphql.duration", duration, tags={"op": name})
if self.execution_context.result.errors:
metrics.increment("graphql.errors", tags={"op": name})
客户端应当为每个查询命名（query Books(...) 而不是匿名查询），否则监控里只有一团匿名流量。这是 GraphQL 生产化的隐形前提。
7.3 部署要点
- 单端点意味着无法按路径做差异化限流，需按 operation 或客户端标识限流
- 关闭 introspection + 开启持久化查询
- 深度限制 + 复杂度限制 + 超时三重兜底
- 查询日志采样存储（完整查询可用于复现与审计）
- 与 REST 共存的过渡期，可用网关按路径分流
GraphQL 与 REST 并非二选一：常见做法是对外保留 REST（利于缓存与生态），对内或对多端提供 GraphQL。选型时值得先读 REST、gRPC 与 GraphQL 的 API 设计对比 ，再结合 GraphQL API 工程化 中的治理经验做决策。
小结
GraphQL 的收益来自「客户端自述字段」，代价是「服务端失去对查询形状的掌控」。因此工程重点必须放在三处：可空性当契约设计（少写不可空、多写描述）、DataLoader 消除 N+1（请求级实例、顺序严格对应）、成本控制三重门（深度限制、复杂度分析、分页上限）。把业务错误放进 Payload 而非 errors、按 operation name 打点，是从「能跑」到「能运维」的分水岭。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。