如何基于FastAPI Strawberry实现嵌套数据结构的GraphQL Mutation

1. 安装依赖 #

先安装所需包：

pip install fastapi strawberry-graphql uvicorn

2. 定义GraphQL输入类型 #

对应请求体的嵌套结构，层级化定义输入类型：

import strawberry
from typing import List, Dict

# 最内层详情输入
@strawberry.input
class DetailInput:
    name: str
    designation: str

# 分组输入（包含详情列表）
@strawberry.input
class GroupInput:
    details: List[DetailInput]

3. 定义GraphQL输出类型 #

匹配响应格式定义输出结构：

# 排名输出项
@strawberry.type
class RankOutput:
    id: str
    rank: int

# 详情输出项
@strawberry.type
class DetailOutput:
    id: str
    name: str
    rank: int

# 最终响应结构
@strawberry.type
class ResponseOutput:
    ranks: List[RankOutput]
    details: List[DetailOutput]

4. 实现Mutation处理逻辑 #

编写嵌套数据解析与响应生成的核心逻辑：

@strawberry.type
class Mutation:
    @strawberry.mutation
    def process_groups(self, groups: Dict[str, GroupInput]) -> ResponseOutput:
        ranks = []
        details = []
        # 按分组ID排序生成排名（适配"0""1"这类有序字符串ID）
        sorted_group_ids = sorted(groups.keys())
        
        for rank, group_id in enumerate(sorted_group_ids, start=1):
            ranks.append(RankOutput(id=group_id, rank=rank))
            # 遍历当前分组的所有详情项
            for detail in groups[group_id].details:
                details.append(DetailOutput(id=group_id, name=detail.name, rank=rank))
        
        return ResponseOutput(ranks=ranks, details=details)

5. 整合到FastAPI #

将GraphQL服务挂载到FastAPI应用：

from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

# 创建GraphQL Schema
schema = strawberry.Schema(mutation=Mutation)

# 初始化FastAPI应用
app = FastAPI()

# 挂载GraphQL路由，默认访问路径为/graphql
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

6. 测试Mutation #

启动服务后，访问http://localhost:8000/graphql，执行以下Mutation测试：

mutation {
  processGroups(
    groups: {
      "0": {
        details: [
          {name: "sam", designation: "student"},
          {name: "raj", designation: "worker"}
        ]
      },
      "1": {
        details: [
          {name: "ram", designation: "employer"},
          {name: "sekhar", designation: "engineer"}
        ]
      }
    }
  ) {
    ranks {
      id
      rank
    }
    details {
      id
      name
      rank
    }
  }
}

返回结果将完全匹配需求格式：

{
  "data": {
    "processGroups": {
      "ranks": [
        {"id": "0", "rank": 1},
        {"id": "1", "rank": 2}
      ],
      "details": [
        {"id": "0", "name": "sam", "rank": 1},
        {"id": "0", "name": "raj", "rank": 1},
        {"id": "1", "name": "ram", "rank": 2},
        {"id": "1", "name": "sekhar", "rank": 2}
      ]
    }
  }
}

关键说明 #

• 用@strawberry.input实现嵌套输入类型，完美对应原REST请求体结构
• 通过Dict[str, GroupInput]接收键为字符串的嵌套分组数据
• 在resolver中按ID排序生成排名，遍历嵌套详情列表组装输出数据

内容的提问来源于stack exchange，提问作者siva