查询成本分析与深度限制
使用深度限制、复杂度评分和持久化查询，保护 API 免受滥用查询影响。
查询成本分析与深度限制 是 CoddyKit 上的免费 FastAPI Backend Development Bootcamp 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 FastAPI Backend Development Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
Why GraphQL Needs Query Guards
REST endpoints have a fixed cost: you call /users/1/posts and the server decides how much work that is. GraphQL flips control to the client. A single endpoint accepts arbitrary nested queries, so a malicious or careless client can ask for something enormous.
Consider a query that walks user -> friends -> friends -> friends .... Each level multiplies the number of resolvers fired. A few hundred bytes of query text can trigger millions of database round-trips.
- Depth limiting caps how deeply nested a query can be.
- Complexity scoring caps the estimated total cost.
- Persisted queries only allow pre-approved query documents.
In this lesson we build all three for a Strawberry + FastAPI service.
The Nested-Query Attack
The classic abuse is a deeply nested, self-referential query. If your schema lets Author resolve posts and Post resolve author, a client can ping-pong between them indefinitely.
Below is the shape of such a query as plain text. Even without running it, notice that every extra posts { author { ... } } level can multiply fan-out at the data layer.
abusive_query = """
query {
author(id: 1) {
posts {
author {
posts {
author {
posts { title }
}
}
}
}
}
}
"""
# A simple depth counter on the raw text gives a rough idea.
depth = abusive_query.count("{")
print(f"Approximate nesting braces: {depth}")Measuring Depth Properly
Counting braces in text is unreliable (string literals, fragments, aliases all break it). The correct way is to walk the parsed AST that graphql-core produces. Each FieldNode with a nested selection set adds one to the depth.
Here is a standalone, dependency-free walker that computes the maximum selection depth of a nested structure. It mirrors how a real validation rule recurses.
def max_depth(node, current=0):
# node is a dict: {"name": str, "selections": [child, ...]}
selections = node.get("selections", [])
if not selections:
return current
return max(max_depth(child, current + 1) for child in selections)
query = {
"name": "author",
"selections": [
{"name": "posts", "selections": [
{"name": "author", "selections": [
{"name": "title", "selections": []}
]}
]}
],
}
print("Max depth:", max_depth(query))A Depth-Limiting Validation Rule
GraphQL exposes a validation phase that runs after parsing but before execution. If any validation rule reports an error, execution never starts, so no resolvers fire. This is exactly where depth limiting belongs.
The graphql-core library ships a ValidationRule base class. We override enter_operation_definition to measure depth and append a GraphQLError when the limit is exceeded.
from graphql import ValidationRule, GraphQLError
def depth_limit_validator(max_allowed: int):
class DepthLimitRule(ValidationRule):
def enter_operation_definition(self, node, *args):
def depth_of(selection_set, level=1):
if selection_set is None:
return level - 1
return max(
(depth_of(getattr(f, "selection_set", None), level + 1)
for f in selection_set.selections),
default=level,
)
found = depth_of(node.selection_set)
if found > max_allowed:
self.report_error(GraphQLError(
f"Query depth {found} exceeds limit of {max_allowed}.",
nodes=node,
))
return DepthLimitRuleWiring the Rule into Strawberry
Strawberry's Schema accepts an extensions list and forwards extra validation rules through its execution config. The cleanest path is the AddValidationRules extension, which injects custom rules into every request.
Because depth validation happens before resolvers run, an over-deep query is rejected with a 400-style GraphQL error and zero database work.
import strawberry
from strawberry.extensions import AddValidationRules
@strawberry.type
class Query:
@strawberry.field
def hello(self) -> str:
return "world"
schema = strawberry.Schema(
query=Query,
extensions=[
AddValidationRules([depth_limit_validator(max_allowed=7)]),
],
)Why Depth Alone Is Not Enough
Depth limiting stops vertical abuse, but it ignores horizontal abuse. A query only 2 levels deep can still be catastrophic:
search(first: 100000)returns a huge list.- Requesting 50 sibling fields at the same level multiplies resolver work.
- Aliasing the same expensive field 100 times bypasses naive per-field caps.
To cover these we need complexity scoring: assign each field a cost, multiply by list-size arguments, and sum the whole tree. Reject when the total crosses a budget.
A Simple Complexity Estimator
The core idea: walk the query tree, give each field a base cost (default 1), and multiply a subtree's cost by any first/limit argument it carries. A list field that fetches 100 items and contains 3 sub-fields costs roughly 100 * 3.
This standalone estimator captures that multiplier logic so you can reason about budgets before plugging it into a framework.
def complexity(node):
base = 1
children = node.get("selections", [])
child_cost = sum(complexity(c) for c in children)
multiplier = node.get("first", 1)
return base + multiplier * child_cost
feed = {
"name": "feed", "first": 100, "selections": [
{"name": "title", "selections": []},
{"name": "comments", "first": 20, "selections": [
{"name": "body", "selections": []},
]},
],
}
score = complexity(feed)
print("Estimated cost:", score)
print("Allowed?", score <= 1000)Per-Field Cost Annotations
Hard-coding cost 1 everywhere is too blunt. Some fields are cheap (a string column) and some are expensive (a fan-out join or an external API call). In production you annotate each field with its own weight.
A common pattern is a metadata dictionary keyed by field name, looked up during scoring. Strawberry lets you attach such metadata via field directives or a custom extension that reads from a registry like the one below.
FIELD_COSTS = {
"id": 0,
"title": 1,
"author": 2, # a join
"recommendations": 25, # an ML call
}
def weighted_cost(node, costs):
own = costs.get(node["name"], 1)
multiplier = node.get("first", 1)
children = sum(weighted_cost(c, costs) for c in node.get("selections", []))
return own + multiplier * children
query = {"name": "feed", "first": 10, "selections": [
{"name": "title", "selections": []},
{"name": "recommendations", "selections": []},
]}
print("Weighted cost:", weighted_cost(query, FIELD_COSTS))Enforcing a Cost Budget as an Extension
To enforce complexity in Strawberry, write a SchemaExtension that hooks on_validate (or on_operation). It computes the score from the parsed document and raises a GraphQLError if the budget is blown, halting before execution.
Keep the budget configurable per client tier: anonymous traffic gets a small budget, authenticated partners get more. The score and budget are great metrics to log and alert on.
from strawberry.extensions import SchemaExtension
from graphql import GraphQLError
class ComplexityLimiter(SchemaExtension):
def __init__(self, *, max_cost: int = 1000):
self.max_cost = max_cost
def on_validate(self):
document = self.execution_context.graphql_document
if document is None:
yield
return
cost = estimate_document_cost(document) # your AST walker
if cost > self.max_cost:
self.execution_context.errors = [
GraphQLError(f"Query cost {cost} exceeds {self.max_cost}.")
]
yieldPersisted Queries: Allowlisting Documents
Depth and complexity limits are heuristics. The strongest control is the persisted query allowlist: clients send only a hash, and the server executes only documents it has pre-registered (usually extracted from your own frontend at build time).
Arbitrary ad-hoc queries are rejected outright, so attackers cannot craft anything new. The client sends {"extensions": {"persistedQuery": {"sha256Hash": "..."}}} and the server resolves it from a store.
import hashlib
# Build-time: register approved documents by their hash.
ALLOWLIST = {}
def register(query_text: str) -> str:
h = hashlib.sha256(query_text.encode()).hexdigest()
ALLOWLIST[h] = query_text
return h
def resolve_persisted(sha256_hash: str) -> str:
if sha256_hash not in ALLOWLIST:
raise ValueError("PersistedQueryNotFound")
return ALLOWLIST[sha256_hash]
hash_a = register("query { me { id name } }")
print("Registered:", hash_a[:12])
print("Resolved:", resolve_persisted(hash_a))Layering the Defenses
These controls are complementary, not alternatives. A hardened FastAPI + Strawberry gateway typically stacks them in this order:
- Persisted queries first — in strict mode, reject anything not on the allowlist before parsing.
- Depth limit — cheap structural guard during validation.
- Complexity budget — weighted cost ceiling, tiered per client.
- Rate limiting — cap requests per token/IP at the FastAPI middleware layer.
Validation runs before resolvers, so depth and complexity rejections cost almost nothing. Always return the computed score in logs so you can tune budgets from real traffic instead of guessing.
Quick Check
A client sends a query that is only 2 levels deep but uses search(first: 50000) and selects 30 sibling fields under each result. Your server has a depth limit of 10 but no complexity scoring. What happens?
Recap
You now have a layered defense against abusive GraphQL queries on FastAPI + Strawberry:
- Depth limiting walks the AST during the validation phase and rejects over-nested queries before any resolver runs.
- Complexity scoring assigns weighted, per-field costs and multiplies by list-size arguments to enforce a budget, catching wide and aliased queries that depth misses.
- Persisted queries allowlist hashed documents so only pre-approved queries execute — the strongest guarantee.
- Stack them with rate limiting, tier budgets per client, and log the computed scores to tune limits from real data.
Remember the key trade-off: depth guards vertical abuse, complexity guards horizontal abuse, and persisted queries eliminate the unknown entirely.
常见问题解答
「查询成本分析与深度限制」课时是免费的吗？
是的 — 「查询成本分析与深度限制」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 FastAPI Backend Development Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。
「查询成本分析与深度限制」这节课中我会学到什么？
使用深度限制、复杂度评分和持久化查询，保护 API 免受滥用查询影响。 你通过在浏览器中直接运行的动手代码来练习 FastAPI Backend Development Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 FastAPI Backend Development Bootcamp 需要有经验吗？
无需任何先前经验。CoddyKit 上的 FastAPI Backend Development Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「查询成本分析与深度限制」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 FastAPI Backend Development Bootcamp 课中编写并运行代码吗？
能。每节 FastAPI Backend Development Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。