随着前端架构的演进，GraphQL 正在逐步替代传统 REST API 成为众多现代 Web 应用的数据交互标准。相较于 REST 接口固定的端点和返回结构，GraphQL 以其单端点、按需取数的特性大幅提升了前端开发效率，但也给数据采集带来了新的挑战 —— 你无法再通过 URL 路径区分接口功能，也不能依赖固定的 JSON 结构解析数据。
本文将从基础原理到实战代码，系统讲解 GraphQL 接口的完整抓取流程，涵盖接口识别、Schema 解析、查询构造、分页处理、反爬绕过等核心环节。
一、GraphQL 与 REST 接口抓取的核心差异
在开始抓取之前，首先要理解两者在架构上的本质区别，这是所有抓取策略的出发点：
表格
| 对比维度 | REST API | GraphQL |
|---|---|---|
| 端点设计 | 多端点，一个资源对应一个 URL（如/users/1、/orders/1） | 单端点，所有请求统一走/graphql路径 |
| 数据控制 | 服务端决定返回结构，易出现数据冗余或不足 | 客户端通过查询语句精确指定返回字段 |
| 请求方式 | 通过 HTTP 方法（GET/POST/PUT/DELETE）区分操作类型 | 统一用 POST（少数支持 GET），通过 query/mutation 区分操作 |
| 关联数据 | 获取关联资源需发起多次请求（N+1 问题） | 单次查询可获取多级嵌套的关联数据 |
| 类型系统 | 无强制类型约束，返回结构依赖文档 | 强类型 Schema，支持内省查询自我描述 |
简单来说，REST 抓取的核心是「找对 URL 和参数」，而 GraphQL 抓取的核心是「写对查询语句」。所有数据都从同一个入口进出，你需要通过构造不同的 query 来获取目标数据。
二、第一步：识别与定位 GraphQL 接口
2.1 典型特征识别
GraphQL 接口有非常鲜明的特征，通过浏览器开发者工具即可快速识别：
- 固定路径特征：接口路径通常包含
/graphql、/api/graphql、/graphql/v1等关键词 - 请求体特征：POST 请求的 JSON body 中包含
query字段，常见配套字段还有variables、operationName - 响应体特征：返回 JSON 固定包含
data顶层字段，错误信息在errors数组中
2.2 快速验证方法
找到疑似端点后，可以发送一个最简查询验证是否为 GraphQL 服务：
bash
运行
curl -X POST https://target.com/graphql \ -H "Content-Type: application/json" \ -d '{"query": "{ __typename }"}'如果返回{"data":{"__typename":"Query"}}，即可确认这是一个有效的 GraphQL 入口。
2.3 常见隐藏场景
部分站点会做路径伪装，需要结合网络面板进一步排查：
- 统一走
/api路径，通过 body 内的字段区分 GraphQL 请求 - 使用 GET 请求，将 query 编码到 URL 参数中
- WebSocket 协议承载 GraphQL 订阅（Subscription）操作
三、第二步：解析 Schema 与接口结构
Schema 是 GraphQL API 的「完整说明书」，定义了所有可查询的字段、类型、参数和关联关系。拿到 Schema 就等于拿到了接口的全部能力清单。
3.1 利用内省查询获取完整 Schema
GraphQL 内置了标准的内省（Introspection）机制，通过发送特定查询即可让服务端返回完整的 Schema 定义：
graphql
query IntrospectionQuery { __schema { queryType { name } mutationType { name } types { name kind fields { name args { name type { name ofType { name } } } type { name kind ofType { name } } } } } }将上述查询发送到目标端点，即可获得全量类型定义。返回结果可以导入 GraphQL Voyager 等工具生成可视化的关系图谱，直观梳理数据结构。
3.2 内省查询被禁用的绕过方案
生产环境中很多服务会关闭内省功能，直接查询会返回introspection is not allowed错误。此时可尝试以下绕过手段：
换行 / 空格注入绕过：针对简单的关键字正则拦截，在
__schema和{之间插入换行符json
{"query": "{ __schema\n { queryType { name } } }"}部分实现只做了单行关键字匹配，换行即可突破检测博客园。
Fragment 分片绕过：将内省字段拆分到片段中
graphql
fragment SchemaFrag on __Schema { queryType { name } } query { ...SchemaFrag }请求方式与 Content-Type 切换：
- 尝试 GET 请求，将 query 放在 URL 参数中
- 将
Content-Type改为application/x-www-form-urlencoded，以表单形式提交 query
字段建议信息利用：即使内省被禁，发送错误字段名时，服务端通常会返回「你是否想找 xxx」的提示，可通过穷举逐步推导可用字段。
3.3 逆向前端请求推导结构
如果以上方法都失效，最稳妥的方式是通过抓包逆向：
- 打开浏览器 DevTools 的 Network 面板，操作页面触发数据加载
- 筛选出所有 GraphQL 请求，逐个查看请求体中的 query 和 variables
- 收集同一业务场景下的所有查询，拼接还原出完整的字段结构
这也是针对私有 GraphQL 接口最常用的分析手段。
四、第三步：核心抓取实现
4.1 基础请求构造
GraphQL 请求本质上就是带特定 body 的 HTTP POST 请求，任何支持 HTTP 的工具都能发送。
curl 方式（调试用）：
bash
运行
curl -X POST https://api.example.com/graphql \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your_token" \ -d '{ "query": "query GetUser($id: ID!) { user(id: $id) { id name email } }", "variables": {"id": "123"} }'Python requests 方式（最常用）：
python
运行
import requests GRAPHQL_URL = "https://api.example.com/graphql" HEADERS = {"Content-Type": "application/json"} query = """ query GetProduct($productId: ID!) { product(id: $productId) { id title price stock category { name } } } """ variables = {"productId": "1001"} response = requests.post( GRAPHQL_URL, headers=HEADERS, json={"query": query, "variables": variables} ) data = response.json()["data"]["product"]4.2 使用专业 GraphQL 客户端
对于复杂场景，可以使用gql库，它内置了 Schema 校验、自动重试、传输层优化等能力：
python
运行
from gql import gql, Client from gql.transport.requests import RequestsHTTPTransport transport = RequestsHTTPTransport( url="https://api.example.com/graphql", headers={"Authorization": "Bearer token"}, use_json=True, ) client = Client(transport=transport, fetch_schema_from_transport=False) query = gql(""" query { products(first: 20) { edges { node { id name price } } } } """) result = client.execute(query)4.3 分页抓取处理
GraphQL 有两种主流分页模式，抓取策略完全不同：
偏移量分页（Offset-based）
和 REST 类似，通过page+limit控制，逻辑简单：
graphql
query { products(page: 3, limit: 50) { items { id name price } totalPages } }游标分页（Cursor-based）
这是 GraphQL 最推荐的分页方式（Relay 风格），通过after游标和first数量翻页，需要循环处理：
python
运行
def crawl_products(): all_products = [] after_cursor = None has_next = True while has_next: query = """ query GetProducts($after: String) { products(first: 50, after: $after) { edges { node { id title price stock } } pageInfo { hasNextPage endCursor } } } """ variables = {"after": after_cursor} resp = requests.post( GRAPHQL_URL, json={"query": query, "variables": variables} ).json() page_data = resp["data"]["products"] for edge in page_data["edges"]: all_products.append(edge["node"]) has_next = page_data["pageInfo"]["hasNextPage"] after_cursor = page_data["pageInfo"]["endCursor"] return all_products核心逻辑是：每次请求后提取endCursor，作为下一次请求的after参数，直到hasNextPage为 false。
4.4 批量查询优化
部分 GraphQL 服务支持批量请求（Batching），可以在一次 HTTP 请求中发送多个查询，大幅减少网络开销：
python
运行
queries = [ {"query": "query { product(id: 1) { name price } }"}, {"query": "query { product(id: 2) { name price } }"}, {"query": "query { product(id: 3) { name price } }"}, ] response = requests.post(GRAPHQL_URL, json=queries) # 返回一个数组，顺序与请求对应 results = response.json()是否支持批量取决于服务端实现，需要自行测试验证。
五、常见反爬机制与应对策略
5.1 查询复杂度限制
GraphQL 的速率限制通常不是按请求数，而是按查询复杂度计算。嵌套层级越深、关联字段越多，单次请求消耗的额度越高。
应对策略：
- 扁平化查询：将深层嵌套的关联查询拆分为多次独立请求
- 减少单次返回字段数：只取必需字段，避免一次性拉取全量属性
- 控制分页大小：不要把
first设置过大，50-100 通常是安全区间
5.2 认证与鉴权
绝大多数业务型 GraphQL 接口都需要身份认证，常见形式：
- Bearer Token：放在
Authorization请求头中，注意 token 过期时间和刷新逻辑 - Cookie + Session：保持会话状态，和普通网页抓取一致
- CSRF Token：部分站点要求额外携带 CSRF 头字段，需要从页面或 Cookie 中提取
5.3 字段级权限控制
不要试图通过内省发现的所有字段都能访问。很多字段会做权限校验，未登录或低权限用户请求会返回 null 或报错。
应对方式：以页面实际加载的查询为准，不要盲目添加内省发现的额外字段。
5.4 频率与行为检测
和 REST API 一样，GraphQL 接口也会有 IP 频率限制、异常请求检测。常规的反爬策略同样适用：
- 合理控制请求间隔，添加随机延迟
- 使用代理池分散 IP
- 模拟正常的请求头和 UA
- 保持和浏览器一致的查询结构，不要随意修改字段顺序和参数
六、完整实战案例：商品列表全量抓取
下面给出一个可直接运行的完整示例，演示如何抓取一个标准 Relay 风格的商品 GraphQL 接口：
python
运行
import requests import time import json class GraphQLScraper: def __init__(self, endpoint, token=None): self.endpoint = endpoint self.headers = { "Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } if token: self.headers["Authorization"] = f"Bearer {token}" def fetch_products(self, category_id, page_size=50): all_products = [] after = None page = 0 while True: query = """ query ProductList($categoryId: ID!, $first: Int, $after: String) { productList(categoryId: $categoryId, first: $first, after: $after) { edges { node { id sku title price originalPrice stock salesCount images { url } } } pageInfo { hasNextPage endCursor } totalCount } } """ variables = { "categoryId": category_id, "first": page_size, "after": after } try: resp = requests.post( self.endpoint, headers=self.headers, json={"query": query, "variables": variables}, timeout=10 ) resp.raise_for_status() data = resp.json() except Exception as e: print(f"请求失败: {e}") time.sleep(3) continue if "errors" in data: print(f"GraphQL 错误: {data['errors']}") break result = data["data"]["productList"] for edge in result["edges"]: all_products.append(edge["node"]) page += 1 print(f"已抓取第 {page} 页，累计 {len(all_products)}/{result['totalCount']} 条") if not result["pageInfo"]["hasNextPage"]: break after = result["pageInfo"]["endCursor"] time.sleep(0.5) # 限速保护 return all_products if __name__ == "__main__": scraper = GraphQLScraper("https://api.example.com/graphql") products = scraper.fetch_products(category_id="100", page_size=50) with open("products.json", "w", encoding="utf-8") as f: json.dump(products, f, ensure_ascii=False, indent=2) print(f"抓取完成，共 {len(products)} 条数据已保存")七、合规与风险提示
- 法律合规：抓取数据前请确认目标网站的服务条款和 robots.txt，不得抓取受保护的个人信息或商业敏感数据，不得用于非法用途。
- 访问压力：控制抓取频率，避免对目标服务造成过大负载，高频大规模抓取可能触发法律追责。
- 数据使用：通过接口获取的数据受版权和数据保护法规约束，二次分发或商用需获得授权。
- 账号安全：使用认证账号抓取时，注意账号风控策略，频繁异常请求可能导致账号封禁。
总结
GraphQL 接口抓取的核心思路可以归纳为三步：定位端点 → 解析结构 → 构造查询。相较于 REST 接口，它的学习门槛稍高，但一旦掌握了 Schema 分析和查询构造方法，抓取效率反而更高 —— 返回结构高度规整，无需做复杂的 HTML 解析，数据一致性更好。
实际工作中，大多数场景下通过抓包复用前端的查询语句是最高效的方式，不需要完整推导整个 Schema。只有在需要批量枚举数据、挖掘隐藏字段等进阶场景下，才需要深入使用内省查询和类型推导。
