使用 GraphQL 过滤 Python 中的 JSON 数据
Using GraphQL to Filter a JSON Data in Python
我在我的 python 项目中使用官方 graphene
包。我有一个来自 运行 Docker 容器检查命令的巨大嵌套 JSON 数据,我想从中获取一些字段。我可以 select 所有这些字段一个一个地添加到字典中,如下所示:
def ContainerConfigs(configs):
data = {}
data['Command'] = configs['Args'][-1]
data['Hostname'] = configs['Config']['Hostname']
data['Image'] = configs['Config']['Image']
data['Distro'] = configs['Config']['Labels']['org.label-schema.name']
data['WorkDir'] = configs['Config']['WorkingDir']
data['IPAddress'] = configs['NetworkSettings']['Networks']['bridge']['IPAddress']
return data
但这是一个弱解决方案。我认为它可以使用 GraphQL 进行优化。没有服务器、请求和响应。我想创建它的 Schema
class 并发送参数(JSON 和查询)并让函数执行该查询并返回结果。像这样:
import graphene
# I need to find this part of the code
# class Schema(..):
# ...
# class Query(...):
# ...
def ContainerConfigs(configs, query):
schema = graphene.Schema(query=Query)
# I need to find a way to pass the configs (Json data) to the Query class to
# execute the query_string on the data
query_string = """
query {
Args
Config {
Hostname
Image
WorkingDir
Label{
org.label-schema.name
}
}
NetworkSettings{
Networks{
bridge{
IPAddress
}
}
}
}
"""
result = schema.execute(query_string)
return result
这不是一个很好的技术匹配:你最终会在 GraphQL 语法中重新创建很多 Docker API,并且它仍然具有相同的 nested-dictionary查询后的布局。
原则上,您可以编写与 graphql.org 上的 Docker API. The schema is required; you can't do a GraphQL query without it, but this basically involves making a copy of the entire Docker API object model in your source code. The other thing to take note of here is that the response to a GraphQL query has the same "shape" as the query (see for example the Execution 页面具有相同布局的 GraphQL 架构。
如果您完成所有这些工作,并且您 运行 您显示的 GraphQL 查询,您会得到 nested-dictionary 响应,例如
result = schema.execute(query_string)
hostname = result.data['Config']['Hostname']
这并不比您已有的更好。 (同样,请参阅石墨烯 Execution 文档页面上的示例。)
如果您想提取这些数据并将其封装在某种对象中,您可以创建一个 class 来为您完成。我可能会这样写:
from collections import namedtuple
class ContainerConfigs(namedtuple('ContainerConfigs', ['command', 'hostname', ...])):
@classmethod
def from_docker_inspect(cls, configs):
return cls(
command=configs['Args'][-1],
hostname=configs['Config']['Hostname'],
...
)
info = ContainerConfigs.from_docker_inspect(configs)
print(f"container hostname (not generally useful): {info.hostname}")
我在我的 python 项目中使用官方 graphene
包。我有一个来自 运行 Docker 容器检查命令的巨大嵌套 JSON 数据,我想从中获取一些字段。我可以 select 所有这些字段一个一个地添加到字典中,如下所示:
def ContainerConfigs(configs):
data = {}
data['Command'] = configs['Args'][-1]
data['Hostname'] = configs['Config']['Hostname']
data['Image'] = configs['Config']['Image']
data['Distro'] = configs['Config']['Labels']['org.label-schema.name']
data['WorkDir'] = configs['Config']['WorkingDir']
data['IPAddress'] = configs['NetworkSettings']['Networks']['bridge']['IPAddress']
return data
但这是一个弱解决方案。我认为它可以使用 GraphQL 进行优化。没有服务器、请求和响应。我想创建它的 Schema
class 并发送参数(JSON 和查询)并让函数执行该查询并返回结果。像这样:
import graphene
# I need to find this part of the code
# class Schema(..):
# ...
# class Query(...):
# ...
def ContainerConfigs(configs, query):
schema = graphene.Schema(query=Query)
# I need to find a way to pass the configs (Json data) to the Query class to
# execute the query_string on the data
query_string = """
query {
Args
Config {
Hostname
Image
WorkingDir
Label{
org.label-schema.name
}
}
NetworkSettings{
Networks{
bridge{
IPAddress
}
}
}
}
"""
result = schema.execute(query_string)
return result
这不是一个很好的技术匹配:你最终会在 GraphQL 语法中重新创建很多 Docker API,并且它仍然具有相同的 nested-dictionary查询后的布局。
原则上,您可以编写与 graphql.org 上的 Docker API. The schema is required; you can't do a GraphQL query without it, but this basically involves making a copy of the entire Docker API object model in your source code. The other thing to take note of here is that the response to a GraphQL query has the same "shape" as the query (see for example the Execution 页面具有相同布局的 GraphQL 架构。
如果您完成所有这些工作,并且您 运行 您显示的 GraphQL 查询,您会得到 nested-dictionary 响应,例如
result = schema.execute(query_string)
hostname = result.data['Config']['Hostname']
这并不比您已有的更好。 (同样,请参阅石墨烯 Execution 文档页面上的示例。)
如果您想提取这些数据并将其封装在某种对象中,您可以创建一个 class 来为您完成。我可能会这样写:
from collections import namedtuple
class ContainerConfigs(namedtuple('ContainerConfigs', ['command', 'hostname', ...])):
@classmethod
def from_docker_inspect(cls, configs):
return cls(
command=configs['Args'][-1],
hostname=configs['Config']['Hostname'],
...
)
info = ContainerConfigs.from_docker_inspect(configs)
print(f"container hostname (not generally useful): {info.hostname}")