扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长
发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/
在 FastAPI 中集成 GraphQL 需要借助第三方库，这里使用 Strawberry（版本要求 strawberry-graphql≥0.215.3）。以下为典型项目结构：
# main.py
import strawberry
from fastapi import FastAPI
from strawberry.asgi import GraphQL
@strawberry.type
class Book:
title: str
author: str
@strawberry.type
class Query:
@strawberry.field
def books(self) -> list[Book]:
return [Book(title="FastAPI进阶", author="李华")]
schema = strawberry.Schema(query=Query)
app = FastAPI()
app.add_route("/graphql", GraphQL(schema))
@strawberry.type 定义数据模型/graphql 访问）# 字段成本定义示例
@strawberry.type
class User:
@strawberry.field(
complexity=lambda info, args: args.get("withPosts", False) * 5 + 1
)
def posts(self, with_posts: bool = False) -> list[Post]:
return database.get_posts()
| 参数 | 类型 | 说明 |
|---|---|---|
| info | GraphQLResolveInfo | 查询上下文信息 |
| args | dict | 字段参数集合 |
# 启用复杂度验证的 Schema 配置
from strawberry.extensions import QueryDepthLimiter
schema = strawberry.Schema(
query=Query,
extensions=[
QueryDepthLimiter(max_depth=5),
MaxComplexityLimiter(max_complexity=100)
]
)
客户端请求 → 解析查询语法树 → 计算字段复杂度 → 比对阈值 → 返回结果/错误
# middleware/rate_limit.py
from fastapi import Request
from redis import Redis
class RateLimiter:
def __init__(self, redis: Redis, capacity=10, refill_rate=1):
self.redis = redis
self.capacity = capacity
self.refill_rate = refill_rate # 令牌/秒
async def check_limit(self, key: str):
current = self.redis.get(key) or 0
if int(current) >= self.capacity:
raise HTTPException(429, "Rate limit exceeded")
self.redis.incr(key)
# 综合限流中间件
async def rate_limit_middleware(request: Request, call_next):
client_ip = request.client.host
complexity = calculate_query_complexity(request.json())
rate_limiter = RateLimiter(
capacity=1000 if complexity | 策略类型 | 适用场景 | 实现复杂度 |
|---|---|---|
| 固定窗口 | 简单接口 | ★☆☆ |
| 滑动窗口 | 精准控制 | ★★☆ |
| 动态调整 | 复杂查询 | ★★★ |
pip install fastapi==0.109.0
pip install strawberry-graphql==0.215.3
pip install redis==4.5.5
# app/main.py
import strawberry
from fastapi import FastAPI
from strawberry.asgi import GraphQL
from middleware.rate_limit import RateLimiter
@strawberry.type
class Analytics:
@strawberry.field(complexity=lambda _, args: args.get("detail_level", 1))
def metrics(self, detail_level: int = 1) -> dict:
return {"visitors": 1000, "conversion": detail_level * 0.2}
schema = strawberry.Schema(
query=Analytics,
extensions=[MaxComplexityLimiter(50)]
)
app = FastAPI()
app.add_middleware(RateLimiter, capacity=1000)
app.add_route("/analytics", GraphQL(schema))
# 合法查询
query {
metrics(detailLevel: 2) {
visitors
}
}
# 非法查询（复杂度超限）
query {
metrics(detailLevel: 50) {
conversion
}
}
# 使用 vegeta 进行压测
echo "POST http://localhost:8000/analytics" | vegeta attack -rate=100 -duration=10s
当客户端收到 "Query is too complex. Maximum allowed complexity: 50" 错误时，应如何调整查询？
答案解析：该错误表明查询总复杂度超过服务端设置阈值。解决方案包括：
产生原因：
解决方案：
# 错误处理中间件示例
@app.exception_handler(ValidationError)
async def handle_validation_error(request, exc):
return JSONResponse(
status_code=400,
content={"detail": "GRAPHQL_VALIDATION_FAILED"}
)
预防建议：
（注：实际部署时建议配置APM监控系统，推荐使用Prometheus+Grafana进行复杂度与流量指标可视化）
余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长
，阅读完整的文章：如何在FastAPI中玩转GraphQL的复杂度与限流？
登录查看全部
参与评论
手机查看
返回顶部