Python 与 GraphQL

10 min中级

Python GraphQL API 开发完整指南:Strawberry + FastAPI、Schema/Query/Mutation/Subscription、DataLoader 与工程实践。

前置知识

学习目标

  • 掌握「1. 历史动机与发展脉络」的核心机制、典型用法与常见陷阱
  • 掌握「2. 形式化定义」的核心机制、典型用法与常见陷阱
  • 掌握「3. 理论推导与原理解析」的核心机制、典型用法与常见陷阱
  • 掌握「4. 代码示例(带详尽注释)」的核心机制、典型用法与常见陷阱
  • 掌握「5. 对比分析」的核心机制、典型用法与常见陷阱

1. 历史动机与发展脉络

GraphQL 由 Facebook 于 2012 年在移动端改造中发明,2015 年公开规范,2018 年由 GraphQL Foundation 管理。动机:移动端网络带宽有限,REST 的“服务端决定响应形状”导致过度获取与多次往返;Facebook 需要客户端按需声明字段。

Python 生态:2016 年 Graphene 发布(code-first,Django/Flask/SQLAlchemy 集成成熟);2019 年 Strawberry 发布(基于 dataclasses 与类型注解,与 FastAPI/Starlette 配合良好);Ariadne 提供 schema-first(先写 SDL 再绑定 resolver)。2020 年代 GraphQL 规范稳定在 October 2021 版本,Python 实现持续跟进(__schema 内省、指令、订阅)。

timeline
    title GraphQL 与 Python 生态
    2012 : Facebook 内部发明 GraphQL
    2015 : GraphQL 规范公开
    2016 : Graphene 发布(Python)
    2018 : GraphQL Foundation 成立
    2019 : Strawberry 发布
    2021 : 规范 October 2021 版
    2024+ : Python 实现与 FastAPI 深度集成

2. 形式化定义

2.1 Schema 与类型系统

GraphQL Schema 用 SDL(Schema Definition Language)描述:type Query、type Mutation、type Subscription 是三种根操作类型;对象类型由字段组成,字段可以有参数(arguments);标量类型(Int、Float、String、Boolean、ID)是叶子;枚举、接口、联合、输入对象补充表达力。

2.2 查询与变更

查询(Query)是只读操作,可并行执行字段;变更(Mutation)是写操作,按顺序执行;订阅(Subscription)通过 WebSocket 推送增量。

2.3 Resolver

每个字段的取值由 resolver 决定:(parent, info, **args) -> value。默认 resolver 按字段名读取父对象属性。Resolver 是 GraphQL 执行模型的执行单元。

2.4 请求示例

query GetUser($id: ID!) {
  user(id: $id) {
    id
    name
    posts {
      title
    }
  }
}

客户端声明需要的字段,服务端只返回这些字段,这就是“按需取数”。

flowchart LR
    A["客户端查询"] --> B["GraphQL 端点 /graphql"]
    B --> C["解析与校验"]
    C --> D["执行 resolver 树"]
    D --> E["按字段聚合响应"]
    E --> F["客户端"]

3. 理论推导与原理解析

3.1 执行模型

GraphQL 执行是“解析→校验→执行”三阶段:解析器把查询字符串变成 AST;校验器对照 Schema 检查类型与参数;执行器从根类型开始,深度优先逐字段调用 resolver。每个字段的 resolver 返回 promise 时可并行(Python 中 async resolver 并发执行)。

3.2 N+1 问题

查询列表用户及其文章时,若每个用户的 resolver 单独查库,会产生 1(用户列表)+ N(每个用户的文章)次查询。DataLoader 通过“批处理 + 去重 + 缓存”把 N 次查询合并为 1-2 次:同一 tick 内收集所有 userId,批量 WHERE id IN (...)。

3.3 查询复杂度

恶意或低效客户端可以请求深层嵌套(如 user -> posts -> comments -> user -> ...)。防护手段:深度限制、数量限制、持久化查询白名单、成本分析(按字段计分)。GraphQL 本身不限制查询形状,安全是服务端责任。

3.4 与 REST 的对比推导

REST 的响应形状由资源表示决定,客户端无法裁剪;GraphQL 把“形状选择权”交给客户端,代价是服务端缓存与监控更复杂(端点唯一,无法按 URL 缓存)。推导结论:数据消费者多样、字段裁剪需求强时 GraphQL 优势明显;简单内部 CRUD 用 REST 更省。

4. 代码示例(带详尽注释)

4.1 Strawberry 基础 Schema

from __future__ import annotations

import strawberry

# 对象类型:映射业务实体
@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    # 解析器字段:根据父对象计算
    @strawberry.field
    def excerpt(self) -> str:
        return self.title[:20]

# 查询根类型:所有只读入口
@strawberry.type
class Query:
    @strawberry.field
    def post(self, id: strawberry.ID) -> Post | None:
        # 实际项目从数据库读取
        return Post(id=id, title="GraphQL 教程")

# 构建 Schema 并输出 SDL
schema = strawberry.Schema(query=Query)
print(schema.as_str())

讲解:Strawberry 用 Python 类型注解直接定义 GraphQL 类型,strawberry.ID 映射标量 ID。excerpt 是带逻辑的字段解析器。schema.as_str() 可以导出 SDL 供前端工具(GraphQL Codegen)使用。

4.2 Mutation 与参数

from __future__ import annotations

import strawberry

# 输入类型:变更的批量参数容器
@strawberry.input
class CreatePostInput:
    title: str
    content: str = ""

@strawberry.type
class Post:
    id: strawberry.ID
    title: str

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_post(self, input: CreatePostInput) -> Post:
        # 校验与持久化(示例省略数据库)
        if not input.title.strip():
            raise ValueError("标题不能为空")
        return Post(id="p-1", title=input.title)

schema = strawberry.Schema(query=Query, mutation=Mutation)

讲解:Mutation 使用输入对象类型聚合参数,返回创建后的对象。业务校验在 resolver 中执行,异常会被 GraphQL 包装进 errors 数组返回。

4.3 接入 FastAPI

from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

import strawberry

@strawberry.type
class Query:
    @strawberry.field
    def hello(self) -> str:
        return "Hello GraphQL"

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

讲解:GraphQLRouter 把 Strawberry Schema 挂载到 FastAPI,自动提供 /graphql 端点与 GraphiQL 调试界面。启动 uvicorn main:app 后可在浏览器中交互查询。

4.4 Ariadne schema-first 方案

from ariadne import QueryType, make_executable_schema, graphql_sync
from ariadne.asgi import GraphQL

# 1. 先用 SDL 定义 Schema
type_defs = """
    type Query {
        greeting(name: String!): String!
    }
"""

# 2. 绑定 resolver 到对应字段
query = QueryType()

@query.field("greeting")
def resolve_greeting(_, info, name: str) -> str:
    return f"Hello, {name}!"

# 3. 组合 schema
schema = make_executable_schema(type_defs, query)

讲解:Ariadne 坚持 schema-first:Schema 是契约(SDL 文件),resolver 用装饰器绑定。适合“前后端先约定 Schema、再并行开发”的团队流程。

4.5 DataLoader 解决 N+1

from __future__ import annotations

from collections import defaultdict

class PostLoader:
    """简单 DataLoader:按批次加载文章,避免 N+1 查询。"""

    def __init__(self, db):
        self.db = db
        self.pending: dict[str, list] = defaultdict(list)

    # 收集阶段:注册需要加载的 user_id
    def load(self, user_id: str):
        self.pending[user_id] = []
        return self

    # 执行阶段:批量查询并分发结果
    async def dispatch(self):
        if not self.pending:
            return
        ids = list(self.pending)
        rows = await self.db.fetch_posts_by_user_ids(ids)  # WHERE user_id IN (...)
        by_user: dict[str, list] = defaultdict(list)
        for row in rows:
            by_user[row["user_id"]].append(row)
        self.pending = defaultdict(list)
        return by_user

讲解:真实项目直接使用 aiodataloader 库:loader.load(user_id) 返回 future,dispatch 在同一事件循环 tick 批量执行。核心思想是“把 N 次查询合并为一次 IN 查询”。

4.6 订阅(Subscription)

import asyncio
import strawberry

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def counter(self, max: int = 5) -> int:
        # 每秒推送一个递增数字
        for i in range(max):
            await asyncio.sleep(1)
            yield i

schema = strawberry.Schema(query=Query, subscription=Subscription)

讲解:订阅用异步生成器逐次 yield 数据,客户端通过 WebSocket 接收。适用于实时通知、进度推送等场景。

4.7 错误处理与自定义格式

from graphql import GraphQLError
import strawberry

@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: str) -> str:
        if id == "404":
            # 抛出 GraphQL 错误:进入 errors 数组,带扩展信息
            raise GraphQLError(
                "用户不存在",
                extensions={"code": "USER_NOT_FOUND", "id": id},
            )
        return f"user-{id}"

讲解:GraphQLError 的 extensions 字段承载业务错误码,客户端可以按 errors[0].extensions.code 分支处理,而不是解析文案。

5. 对比分析

5.1 GraphQL 与 REST

维度GraphQLREST
端点单一 /graphql资源化多端点
响应形状客户端决定服务端决定
取数效率按需,防过度获取可能过度/不足获取
缓存需自定义HTTP 缓存成熟
版本管理Schema 演进URL 版本化
学习成本较高较低

5.2 Graphene vs Strawberry vs Ariadne

库风格特点适用
Graphenecode-first生态成熟,Django 集成强存量 Django
Strawberrycode-first类型注解现代,FastAPI 友好新项目/异步
Ariadneschema-firstSDL 契约驱动前后端契约协作

5.3 查询与变更

查询可并行解析(字段间无依赖时),变更必须串行(保证写顺序)。设计 API 时,读操作放 Query,写操作放 Mutation,即使 Mutation 只读也要遵循该约定,保证客户端语义清晰。

6. 常见陷阱与最佳实践

陷阱一:resolver 中同步阻塞调用数据库,阻塞事件循环。最佳实践:使用异步驱动(asyncpg、aiosqlite)与 async resolver。

陷阱二:忽略 N+1,列表查询性能崩塌。最佳实践:DataLoader 或预加载(selectinload)。

陷阱三:不限制查询复杂度,被深层嵌套查询打垮。最佳实践:深度/数量限制 + 成本分析。

陷阱四:把业务逻辑写进 resolver。最佳实践:resolver 薄层,逻辑进 service 层,便于测试复用。

陷阱五:Mutation 返回类型不包含变更结果,客户端需要再次查询。最佳实践:返回变更后的对象或结果封装。

陷阱六:错误全部返回 200 + errors 数组,HTTP 层监控失真。最佳实践:按错误类型设置响应状态(或依赖 errors 扩展码)并记录日志指标。

陷阱七:Schema 变更破坏客户端。最佳实践:deprecation 注解、版本化演进、Schema Registry。

7. 工程实践

7.1 项目结构

app/
  schema/
    query.py        # 根查询
    mutation.py     # 根变更
    types.py        # 对象/输入类型
  resolvers/
    user_resolver.py
    post_resolver.py
  services/
    user_service.py  # 业务逻辑
  main.py            # FastAPI + GraphQLRouter

讲解:类型、resolver、业务逻辑三层分离。resolver 只做“取参→调 service→映射结果”,service 可独立单测。

7.2 权限指令

from strawberry.permission import BasePermission

class IsAuthenticated(BasePermission):
    # 权限校验失败时抛出错误
    message = "需要登录"

    async def has_permission(self, source, info, **kwargs) -> bool:
        request = info.context["request"]
        return request.user.is_authenticated

@strawberry.type
class Query:
    @strawberry.field(permission_classes=[IsAuthenticated])
    def profile(self) -> str:
        return "私密资料"

讲解:Strawberry 的权限类装饰字段,在 resolver 执行前校验。权限逻辑集中在类中,可复用、可测试。

7.3 持久化查询与缓存

# 持久化查询:服务端保存哈希与查询的映射
PERSISTED = {
    "sha256hash1": "query Home { posts { title } }",
}

async def resolve_operation(operation_name, query):
    # 客户端传哈希时从白名单取查询
    return PERSISTED.get(operation_name, query)

讲解:持久化查询把查询白名单化,天然限制查询形状与复杂度,同时支持 CDN 缓存(GET + 哈希)。安全性与性能双收。

8. 案例研究:博客平台 GraphQL API

需求:用户、文章、评论三个实体;查询文章列表与详情;发表评论;文章更新时推送通知。

from __future__ import annotations

import strawberry

@strawberry.type
class User:
    id: strawberry.ID
    name: str

@strawberry.type
class Comment:
    id: strawberry.ID
    content: str
    author: User

@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    comments: list[Comment]

    # 字段参数:分页评论
    @strawberry.field
    def comments_page(self, limit: int = 10, offset: int = 0) -> list[Comment]:
        return self.comments[offset:offset + limit]

@strawberry.type
class Query:
    @strawberry.field
    def posts(self) -> list[Post]:
        # DataLoader + service 查询
        return fetch_all_posts()

    @strawberry.field
    def post(self, id: strawberry.ID) -> Post | None:
        return fetch_post(id)

@strawberry.type
class Mutation:
    @strawberry.mutation
    async def add_comment(self, post_id: strawberry.ID, content: str) -> Comment:
        if not content.strip():
            raise ValueError("评论不能为空")
        comment = await create_comment(post_id, content)
        # 触发订阅通知
        await notify_comment(comment)
        return comment

schema = strawberry.Schema(query=Query, mutation=Mutation, subscription=Subscription)

讲解:案例覆盖类型建模、字段参数分页、mutation 校验、订阅通知四大模块。comments_page 演示字段级参数——同一个 Post 类型在不同查询中返回不同评论子集,这是 GraphQL 表达力的典型体现。

9. 知识要点总结与深入讲解

GraphQL 的核心是“Schema 即契约”:类型系统同时服务服务端执行与客户端代码生成。理解 Schema、Resolver、执行模型三者,就掌握了 GraphQL 的主线。

性能关键在数据加载:N+1 是 GraphQL 的先天挑战,DataLoader 是标准解。查询复杂度防护是生产必需,不是可选项。

Python 实现的选择取决于团队工作流:喜欢类型注解与异步用 Strawberry,存量 Django 用 Graphene,契约先行用 Ariadne。三者共享 GraphQL 规范,迁移成本可控。

什么是 GraphQL

GraphQL 是一种 API 查询语言,由 Facebook 开发。与 REST API 不同,GraphQL 让客户端自己决定需要哪些数据,不多不少。在 REST 中,一个接口返回固定的字段,你可能获取了太多不需要的数据,或者需要调用多个接口才能获取足够的数据。GraphQL 用一个接口解决了这个问题。

GraphQL 的核心思想是:客户端描述需要的数据结构,服务端返回完全匹配的数据。这避免了过度获取(获取了不需要的字段)和不足获取(需要调用多个接口)的问题。

基础概念

Schema(模式)

Schema 定义了 API 的类型系统,描述了有哪些数据类型、每个类型有哪些字段、支持哪些查询和变更。Schema 是客户端和服务端之间的契约。

Query(查询)

Query 是读取数据的操作,类似于 REST 的 GET 请求。客户端在 Query 中指定需要的字段,服务端只返回这些字段。

Mutation(变更)

Mutation 是修改数据的操作,类似于 REST 的 POST/PUT/DELETE 请求。用于创建、更新、删除数据。

Resolver(解析器)

Resolver 是服务端处理每个字段的函数。当客户端请求某个字段时,对应的 Resolver 被调用来获取数据。

Type(类型)

GraphQL 有内置的标量类型(Int、Float、String、Boolean、ID),也支持自定义对象类型。每个类型有一组字段,字段可以有参数。

快速上手

安装

# 安装 Strawberry GraphQL(现代的 Python GraphQL 库)
pip install strawberry-graphql fastapi

# 或者安装 Graphene(老牌 GraphQL 库)
# pip install graphene

最简单的 GraphQL API

# app.py
import strawberry
from strawberry.fastapi import GraphQLRouter
from fastapi import FastAPI

# 定义 GraphQL 类型
@strawberry.type
class User:
    name: str
    age: int
    email: str

# 模拟数据
users_db = [
    User(name="张三", age=25, email="zhangsan@example.com"),
    User(name="李四", age=30, email="lisi@example.com"),
]

# 定义 Query(查询操作)
@strawberry.type
class Query:
    @strawberry.field
    def users(self) -> list[User]:
        """获取所有用户"""
        return users_db

    @strawberry.field
    def user(self, name: str) -> User | None:
        """按名字查询用户"""
        for u in users_db:
            if u.name == name:
                return u
        return None

# 创建 GraphQL Schema
schema = strawberry.Schema(query=Query)

# 集成到 FastAPI
app = FastAPI()
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

运行:

uvicorn app:app --reload

访问 http://localhost:8000/graphql 可以打开交互式查询界面(GraphQL Playground)。

查询示例

客户端发送以下查询:

# 获取所有用户,但只要 name 和 age 字段
query {
  users {
    name
    age
  }
}

服务端返回:

{
  "data": {
    "users": [
      { "name": "张三", "age": 25 },
      { "name": "李四", "age": 30 }
    ]
  }
}

注意:虽然 User 类型有 email 字段,但因为查询中没有请求 email,所以返回数据中不包含它。

# 按名字查询用户
query {
  user(name: "张三") {
    name
    email
  }
}

详细用法

定义 Mutation(数据变更)

import strawberry
from strawberry.fastapi import GraphQLRouter
from fastapi import FastAPI

@strawberry.type
class Article:
    id: int
    title: str
    content: str
    author: str

articles_db: list[Article] = []
next_id = 1

@strawberry.type
class Query:
    @strawberry.field
    def articles(self) -> list[Article]:
        """获取所有文章"""
        return articles_db

    @strawberry.field
    def article(self, article_id: int) -> Article | None:
        """按 ID 获取文章"""
        for a in articles_db:
            if a.id == article_id:
                return a
        return None

# 定义输入类型(用于 Mutation 的参数)
@strawberry.input
class ArticleInput:
    title: str
    content: str
    author: str

# 定义 Mutation
@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_article(self, input: ArticleInput) -> Article:
        """创建文章"""
        global next_id
        article = Article(
            id=next_id,
            title=input.title,
            content=input.content,
            author=input.author,
        )
        articles_db.append(article)
        next_id += 1
        return article

    @strawberry.mutation
    def delete_article(self, article_id: int) -> bool:
        """删除文章"""
        for i, a in enumerate(articles_db):
            if a.id == article_id:
                articles_db.pop(i)
                return True
        return False

schema = strawberry.Schema(query=Query, mutation=Mutation)
app = FastAPI()
app.include_router(GraphQLRouter(schema), prefix="/graphql")

Mutation 查询示例:

# 创建文章
mutation {
  createArticle(
    input: { title: "GraphQL 入门", content: "这是一篇关于 GraphQL 的教程", author: "张三" }
  ) {
    id
    title
  }
}

# 删除文章
mutation {
  deleteArticle(articleId: 1)
}

关联类型

import strawberry

@strawberry.type
class Author:
    id: int
    name: str
    email: str

@strawberry.type
class Book:
    id: int
    title: str
    author_id: int

    # 关联字段:通过 Resolver 获取作者信息
    @strawberry.field
    def author(self, info) -> Author:
        """获取书籍的作者"""
        return get_author_by_id(self.author_id)

authors_db = [
    Author(id=1, name="张三", email="zhangsan@example.com"),
    Author(id=2, name="李四", email="lisi@example.com"),
]

books_db = [
    Book(id=1, title="Python 入门", author_id=1),
    Book(id=2, title="GraphQL 实战", author_id=1),
    Book(id=3, title="Web 开发指南", author_id=2),
]

def get_author_by_id(author_id: int) -> Author:
    for a in authors_db:
        if a.id == author_id:
            return a
    raise ValueError(f"作者 ID {author_id} 不存在")

查询关联数据:

# 获取书籍及其作者信息
query {
  books {
    title
    author {
      name
      email
    }
  }
}

枚举类型

import strawberry

@strawberry.enum
class Status:
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"

@strawberry.type
class Post:
    id: int
    title: str
    status: Status

@strawberry.type
class Query:
    @strawberry.field
    def posts(self, status: Status | None = None) -> list[Post]:
        """按状态过滤文章"""
        if status:
            return [p for p in posts_db if p.status == status]
        return posts_db

分页查询

import strawberry
from typing import Generic, TypeVar

T = TypeVar("T")

# 通用分页类型
@strawberry.type
class Pagination(Generic[T]):
    items: list[T]
    total: int
    page: int
    page_size: int

@strawberry.type
class Query:
    @strawberry.field
    def articles(self, page: int = 1, page_size: int = 10) -> Pagination[Article]:
        """分页获取文章"""
        start = (page - 1) * page_size
        end = start + page_size
        items = articles_db[start:end]
        return Pagination(
            items=items,
            total=len(articles_db),
            page=page,
            page_size=page_size,
        )

常见场景

与 SQLAlchemy 集成

import strawberry
from strawberry.fastapi import GraphQLRouter
from sqlalchemy import select
from sqlalchemy.orm import Session

# 将 SQLAlchemy 模型转换为 GraphQL 类型
@strawberry.type
class UserType:
    id: int
    name: str
    email: str

    @classmethod
    def from_orm(cls, user):
        """从 ORM 模型创建 GraphQL 类型"""
        return cls(id=user.id, name=user.name, email=user.email)

@strawberry.type
class Query:
    @strawberry.field
    def users(self, info) -> list[UserType]:
        """获取所有用户"""
        db: Session = info.context["db"]
        users = db.execute(select(User)).scalars().all()
        return [UserType.from_orm(u) for u in users]

    @strawberry.field
    def user(self, info, user_id: int) -> UserType | None:
        """按 ID 获取用户"""
        db: Session = info.context["db"]
        user = db.get(User, user_id)
        return UserType.from_orm(user) if user else None

注意事项与常见错误

N+1 查询问题

GraphQL 的关联查询容易产生 N+1 问题。比如查询 10 本书及其作者,可能产生 1 次查书 + 10 次查作者 = 11 次数据库查询。解决方法是使用 DataLoader 批量加载。

查询深度限制

恶意客户端可能发送极深层嵌套的查询,消耗服务器资源。应该限制查询深度:

from strawberry.extensions import QueryDepthLimiter

schema = strawberry.Schema(
    query=Query,
    extensions=[QueryDepthLimiter(max_depth=5)]
)

不要把 GraphQL 当数据库

GraphQL 是 API 层,不是数据库。不要把数据库表结构直接暴露为 GraphQL 类型。应该设计面向客户端的 API 类型,在 Resolver 中做数据转换。

进阶用法

使用 DataLoader 批量加载

pip install strawberry-graphql
from strawberry.dataloader import DataLoader

async def load_authors(author_ids: list[int]) -> list[Author]:
    """批量加载作者"""
    authors = await batch_get_authors(author_ids)
    return [next((a for a in authors if a.id == aid), None) for aid in author_ids]

author_loader = DataLoader(load_authors)

@strawberry.type
class Book:
    id: int
    title: str
    author_id: int

    @strawberry.field
    async def author(self, info) -> Author:
        """使用 DataLoader 批量加载作者"""
        return await info.context["author_loader"].load(self.author_id)

订阅(Subscription)

GraphQL 订阅支持实时数据推送:

import asyncio
import strawberry

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def message_added(self, room_id: str) -> str:
        """订阅新消息"""
        while True:
            message = await get_new_message(room_id)
            if message:
                yield message
            await asyncio.sleep(1)