GraphQL 教程
FastAPI + GraphQL 教程
目录
- 介绍
- GraphQL 基础
- 查询操作
- 变更操作
- 高级特性
- 最佳实践
介绍
GraphQL 是一种用于 API 的查询语言,它提供了一种更高效、强大和灵活的替代 REST 的方案。本教程将介绍如何在 FastAPI 项目中使用 GraphQL。
为什么选择 GraphQL?
-
按需获取数据
- 客户端可以精确指定需要的数据
- 避免过度获取和数据不足的问题
- 减少网络传输量
-
单个请求获取多个资源
- 减少 HTTP 请求次数
- 提高应用性能
- 简化前端逻辑
-
强类型系统
- 编译时错误检查
- 自动生成文档
- 更好的开发体验
GraphQL 基础
1. Schema 定义
@strawberry.type
class User:
id: int
username: str
email: str
is_active: bool
@strawberry.type
class Query:
@strawberry.field
def user(self, id: int) -> Optional[User]:
return get_user(id)
@strawberry.field
def users(self) -> List[User]:
return get_users()
2. 类型系统
# 标量类型
Int: 整数
Float: 浮点数
String: 字符串
Boolean: 布尔值
ID: 唯一标识符
# 对象类型
type User {
id: ID!
username: String!
email: String!
isActive: Boolean!
}
查询操作
1. 基本查询
# 查询单个用户
query {
user(id: 1) {
id
username
email
}
}
# 查询用户列表
query {
users {
id
username
email
isActive
}
}
2. 带参数的查询
# 带过滤条件的查询
query {
users(filter: {
isActive: true
role: "admin"
}) {
id
username
email
}
}
3. 查询片段
fragment UserFields on User {
id
username
email
}
query {
user(id: 1) {
...UserFields
isActive
}
}
变更操作
1. 创建资源
mutation {
createUser(
username: "newuser"
email: "new@example.com"
password: "password123"
) {
id
username
email
}
}
2. 更新资源
mutation {
updateUser(
id: 1
data: {
username: "updateduser"
email: "updated@example.com"
}
) {
id
username
email
}
}
3. 删除资源
mutation {
deleteUser(id: 1) {
success
message
}
}
高级特性
1. 批量操作
mutation {
createUsers(
users: [
{ username: "user1", email: "user1@example.com" }
{ username: "user2", email: "user2@example.com" }
]
) {
id
username
}
}
2. 嵌套查询
query {
users {
id
username
posts {
id
title
comments {
id
content
}
}
}
}
3. 订阅
subscription {
userCreated {
id
username
email
}
}
最佳实践
1. 查询优化
- 使用片段复用字段
fragment UserFields on User {
id
username
email
}
query {
user1: user(id: 1) {
...UserFields
}
user2: user(id: 2) {
...UserFields
}
}
- 限制查询深度
@strawberry.field
def users(self, limit: int = 10, offset: int = 0) -> List[User]:
return get_users(limit=limit, offset=offset)
2. N+1 问题解决
from strawberry.dataloader import DataLoader
async def load_posts_by_user_ids(keys: List[int]) -> List[List[Post]]:
posts = await get_posts_for_users(keys)
return group_posts_by_user(posts)
@strawberry.field
async def posts(self, info: Info) -> List[Post]:
loader = info.context["posts_loader"]
return await loader.load(self.id)
3. 错误处理
@strawberry.mutation
def create_user(self, username: str, email: str) -> Union[User, Error]:
try:
user = create_user_in_db(username, email)
return user
except ValidationError as e:
return Error(message=str(e))
4. 认证和授权
@strawberry.field
def protected_data(self, info: Info) -> str:
user = info.context.get("user")
if not user:
raise PermissionError("Not authenticated")
if not user.is_admin:
raise PermissionError("Not authorized")
return "sensitive data"
使用工具
1. GraphQL Playground
- 访问 http://localhost:8899/graphql
- 交互式文档
- 自动补全
- 实时验证
2. curl 请求
curl -X POST \
-H "Content-Type: application/json" \
-d '{"query": "{ users { id username } }"}' \
http://localhost:8899/graphql
3. Python 请求
import requests
def query_graphql():
query = """
{
users {
id
username
email
}
}
"""
response = requests.post(
'http://localhost:8899/graphql',
json={'query': query}
)
return response.json()
调试技巧
- 使用 __typename
query {
users {
__typename
id
username
}
}
- 查询架构信息
query {
__schema {
types {
name
fields {
name
type {
name
}
}
}
}
}
- 使用别名
query {
activeUsers: users(filter: { isActive: true }) {
id
username
}
inactiveUsers: users(filter: { isActive: false }) {
id
username
}
}