如何借助GraphQLObjectType将GraphQLResponse字典反序列化为Python对象?
解决方案
针对你的需求,这里提供两种可行的反序列化方案,分别适配简单场景和复杂嵌套场景:
1. 手动映射(适合结构简单的响应)
如果你的GraphQL返回结构不复杂,可以直接定义对应Python类(推荐用dataclass简化代码),然后手动将字典字段映射到类属性:
步骤1:定义对应Python类
from dataclasses import dataclass from typing import Optional, List @dataclass class Post: title: str content: str published: bool @dataclass class User: id: str name: str email: Optional[str] posts: List[Post]
步骤2:编写反序列化函数
def deserialize_user(user_data: dict) -> User: # 可选:基于GraphQLObjectType验证字段完整性 # expected_fields = your_object_type.fields.keys() # if not all(f in user_data for f in expected_fields): # raise ValueError("响应数据缺少schema定义的字段") # 递归处理嵌套的Post列表 posts = [Post(**post) for post in user_data.get("posts", [])] return User( id=user_data["id"], name=user_data["name"], email=user_data.get("email"), posts=posts )
使用时直接传入响应字典:
# 假设response是aiographql-client返回的GraphQLResponse user_obj = deserialize_user(response.data["user"])
2. 基于schema自动反序列化(适合复杂嵌套结构)
如果响应包含多层嵌套对象或大量字段,可以利用graphql-core的GraphQLObjectType元数据,动态生成Python类并完成反序列化:
通用反序列化实现
from dataclasses import make_dataclass from typing import List, Any from graphql import ( GraphQLObjectType, GraphQLList, GraphQLNonNull, GraphQLScalarType, GraphQLID, GraphQLString, GraphQLInt, GraphQLFloat, GraphQLBoolean ) # 映射GraphQL标量到Python原生类型 _SCALAR_TYPE_MAP = { GraphQLID.name: str, GraphQLString.name: str, GraphQLInt.name: int, GraphQLFloat.name: float, GraphQLBoolean.name: bool } def _get_python_type(graphql_type): # 处理NonNull包装器 while isinstance(graphql_type, GraphQLNonNull): graphql_type = graphql_type.of_type if isinstance(graphql_type, GraphQLObjectType): # 嵌套对象,递归生成类型 return _generate_dataclass(graphql_type) elif isinstance(graphql_type, GraphQLList): item_type = _get_python_type(graphql_type.of_type) return List[item_type] elif isinstance(graphql_type, GraphQLScalarType): return _SCALAR_TYPE_MAP.get(graphql_type.name, str) raise TypeError(f"不支持的GraphQL类型:{graphql_type}") def _generate_dataclass(obj_type: GraphQLObjectType): # 根据GraphQLObjectType生成dataclass fields = [] for field_name, field_def in obj_type.fields.items(): py_type = _get_python_type(field_def.type) fields.append((field_name, py_type)) return make_dataclass(obj_type.name, fields) def deserialize(data: dict, obj_type: GraphQLObjectType) -> Any: # 递归反序列化字典到Python对象 cls = _generate_dataclass(obj_type) kwargs = {} for field_name, field_def in obj_type.fields.items(): value = data.get(field_name) if value is None: kwargs[field_name] = None continue field_type = field_def.type while isinstance(field_type, GraphQLNonNull): field_type = field_type.of_type if isinstance(field_type, GraphQLObjectType): kwargs[field_name] = deserialize(value, field_type) elif isinstance(field_type, GraphQLList): item_type = field_type.of_type while isinstance(item_type, GraphQLNonNull): item_type = item_type.of_type if isinstance(item_type, GraphQLObjectType): kwargs[field_name] = [deserialize(item, item_type) for item in value] else: scalar_type = _SCALAR_TYPE_MAP.get(item_type.name, str) kwargs[field_name] = [scalar_type(item) for item in value] else: scalar_type = _SCALAR_TYPE_MAP.get(field_type.name, str) kwargs[field_name] = scalar_type(value) return cls(**kwargs)
使用示例
# 假设你已经获取了目标GraphQLObjectType(比如从schema中解析得到) # 假设response.data是返回的原始字典 deserialized_obj = deserialize(response.data["your_query_key"], target_object_type) # 直接访问对象属性 print(deserialized_obj.id) print(deserialized_obj.nested_field.sub_field)
扩展说明
- 自定义标量:如果你的schema包含自定义标量,只需要在
_SCALAR_TYPE_MAP中添加对应的类型映射即可。 - 类缓存:如果需要重复使用同一类型的对象,可以添加一个缓存字典,避免重复生成dataclass。
内容的提问来源于stack exchange,提问作者nettrino
相关产品推荐
相关产品推荐

