如何解析 yaml 列表中的嵌套字典?
How to parse nested dictionary inside a list in yaml?
我正在解析 YAML 文件以搜索任意键的值。目前我可以在第一级解析任何字典,但不能解析嵌套字典。
我尝试修改 中的示例以解析列表中的字典,但这会导致错误:
AttributeError: 'CommentedSeq' object has no attribute 'items'
当查看 http://yaml-online-parser.appspot.com/ 的规范输出时,它显示有一张地图和一个序列,我一直无法解释。
未修改的解析函数不会输出任何错误,但是它在列表中看不到任何内容。
修改解析函数returns上面的AttributeError
示例 YAML 文件:https://pastebin.com/BhwyPa7V
完整项目:https://github.com/Just-Insane/helm-vault/blob/master/vault.py
解析函数(未修改):
def dict_walker(node, pattern, path=None):
path = path if path is not None else ""
for key, value in node.items():
if isinstance(value, dict):
dict_walker(value, pattern=pattern, path=f"{path}/{key}")
elif value == pattern:
if action == "enc":
node[key] = input(f"Input a value for {path}/{key}: ")
vault_write(node[key], path, key)
elif (action == "dec") or (action == "view") or (action == "edit"):
value = vault_read(path, key)
node[key] = value
解析函数(修改):
def dict_walker(node, pattern, path=None):
path = path if path is not None else ""
for key, value in node.items():
if isinstance(value, dict):
dict_walker(value, pattern=pattern, path=f"{path}/{key}")
elif isinstance(value, list):
for item in value:
for value in dict_walker(value, pattern=pattern, path=f"{path}/{key}"):
if value == pattern:
if action == "enc":
node[key] = input(f"Input a value for {path}/{key}: ")
vault_write(node[key], path, key)
elif (action == "dec") or (action == "view") or (action == "edit"):
value = vault_read(path, key)
node[key] = value
elif value == pattern:
if action == "enc":
node[key] = input(f"Input a value for {path}/{key}: ")
vault_write(node[key], path, key)
elif (action == "dec") or (action == "view") or (action == "edit"):
value = vault_read(path, key)
node[key] = value
预期结果:
嵌套字典解析成功,可以修改里面的值。
实际结果:
使用未修改的代码,列表中的值根本看不到。
使用修改后的代码,存在CommentedSeq
导致的属性错误。目前还不清楚为什么它没有被解析为列表。
正如我在您链接的答案中指出的那样,解析已完全完成
甚至在 yaml.load()
方法 returns 之前。你所做的是
遍历加载的数据。
您的 dict_walker()
基于 find()
函数,该函数仅适用于
来自 the question 的无趣 YAML 输入
我回答了。它假定 YAML:
only consists of (plain) scalars that are strings, mappings, and mapping keys that are scalars.
那里介绍的lookup
函数可以处理序列,所以就是
你的 dict_walker()
功能需要基于什么(以及
你拥有的功能有一个合适的名字:它只能走过去
在节点上调用 .items()
方法的字典假设 node
是 dict
).
假设文件 input.yaml
中的 YAML 示例
完成树并到达嵌套在列表中的字典(从嵌套的映射创建
在 YAML 中的序列内):
import sys
from pathlib import Path
import ruamel.yaml
in_file = Path('input.yaml')
action = "view"
def vault_read(path, key):
# dummy function to show functionality
vault = {
("/spec/acme", "email"): "repl_0",
("/spec/acme/dns01/providers/0/cloudflare", "email"): "repl_1",
("/spec/acme/dns01/providers/0/cloudflare/apiKeySecretRef", "key"): 42,
}
return vault.get((path, key), "not found")
def walk_data(node, pattern, path=None):
if path is None:
path = ""
if isinstance(node, dict):
for key, value in node.items():
if value == pattern:
if action == "enc":
node[key] = input(f"Input a value for {path}/{key}: ")
vault_write(node[key], path, key)
elif (action == "dec") or (action == "view") or (action == "edit"):
node[key] = vault_read(path, key)
else:
walk_data(value, pattern, path=f"{path}/{key}")
elif isinstance(node, list):
for idx, item in enumerate(node):
walk_data(item, pattern, path=f"{path}/{idx}")
yaml = ruamel.yaml.YAML()
data = yaml.load(in_file)
walk_data(data, "changeme")
yaml.dump(data, sys.stdout)
给出:
apiVersion: certmanager.k8s.io/v1alpha1
kind: ClusterIssuer
metadata:
name: letsencrypt-production
spec:
acme:
# You must replace this email address with your own.
# Let's Encrypt will use this to contact you about expiring
# certificates, and issues related to your account.
email: repl_0
server: https://acme-v02.api.letsencrypt.org/directory
privateKeySecretRef:
# Secret resource used to store the account's private key.
name: letsencrypt-production
# Enable the HTTP01 challenge mechanism for this Issuer
dns01:
providers:
- name: prod-cloudflare
cloudflare:
email: repl_1
apiKeySecretRef:
name: cloudflare-api-key-secret
key: 42
我正在解析 YAML 文件以搜索任意键的值。目前我可以在第一级解析任何字典,但不能解析嵌套字典。
我尝试修改 中的示例以解析列表中的字典,但这会导致错误:
AttributeError: 'CommentedSeq' object has no attribute 'items'
当查看 http://yaml-online-parser.appspot.com/ 的规范输出时,它显示有一张地图和一个序列,我一直无法解释。
未修改的解析函数不会输出任何错误,但是它在列表中看不到任何内容。
修改解析函数returns上面的AttributeError
示例 YAML 文件:https://pastebin.com/BhwyPa7V
完整项目:https://github.com/Just-Insane/helm-vault/blob/master/vault.py
解析函数(未修改):
def dict_walker(node, pattern, path=None):
path = path if path is not None else ""
for key, value in node.items():
if isinstance(value, dict):
dict_walker(value, pattern=pattern, path=f"{path}/{key}")
elif value == pattern:
if action == "enc":
node[key] = input(f"Input a value for {path}/{key}: ")
vault_write(node[key], path, key)
elif (action == "dec") or (action == "view") or (action == "edit"):
value = vault_read(path, key)
node[key] = value
解析函数(修改):
def dict_walker(node, pattern, path=None):
path = path if path is not None else ""
for key, value in node.items():
if isinstance(value, dict):
dict_walker(value, pattern=pattern, path=f"{path}/{key}")
elif isinstance(value, list):
for item in value:
for value in dict_walker(value, pattern=pattern, path=f"{path}/{key}"):
if value == pattern:
if action == "enc":
node[key] = input(f"Input a value for {path}/{key}: ")
vault_write(node[key], path, key)
elif (action == "dec") or (action == "view") or (action == "edit"):
value = vault_read(path, key)
node[key] = value
elif value == pattern:
if action == "enc":
node[key] = input(f"Input a value for {path}/{key}: ")
vault_write(node[key], path, key)
elif (action == "dec") or (action == "view") or (action == "edit"):
value = vault_read(path, key)
node[key] = value
预期结果:
嵌套字典解析成功,可以修改里面的值。
实际结果:
使用未修改的代码,列表中的值根本看不到。
使用修改后的代码,存在
CommentedSeq
导致的属性错误。目前还不清楚为什么它没有被解析为列表。
正如我在您链接的答案中指出的那样,解析已完全完成
甚至在 yaml.load()
方法 returns 之前。你所做的是
遍历加载的数据。
您的 dict_walker()
基于 find()
函数,该函数仅适用于
来自 the question 的无趣 YAML 输入
我回答了。它假定 YAML:
only consists of (plain) scalars that are strings, mappings, and mapping keys that are scalars.
那里介绍的lookup
函数可以处理序列,所以就是
你的 dict_walker()
功能需要基于什么(以及
你拥有的功能有一个合适的名字:它只能走过去
在节点上调用 .items()
方法的字典假设 node
是 dict
).
假设文件 input.yaml
中的 YAML 示例
完成树并到达嵌套在列表中的字典(从嵌套的映射创建
在 YAML 中的序列内):
import sys
from pathlib import Path
import ruamel.yaml
in_file = Path('input.yaml')
action = "view"
def vault_read(path, key):
# dummy function to show functionality
vault = {
("/spec/acme", "email"): "repl_0",
("/spec/acme/dns01/providers/0/cloudflare", "email"): "repl_1",
("/spec/acme/dns01/providers/0/cloudflare/apiKeySecretRef", "key"): 42,
}
return vault.get((path, key), "not found")
def walk_data(node, pattern, path=None):
if path is None:
path = ""
if isinstance(node, dict):
for key, value in node.items():
if value == pattern:
if action == "enc":
node[key] = input(f"Input a value for {path}/{key}: ")
vault_write(node[key], path, key)
elif (action == "dec") or (action == "view") or (action == "edit"):
node[key] = vault_read(path, key)
else:
walk_data(value, pattern, path=f"{path}/{key}")
elif isinstance(node, list):
for idx, item in enumerate(node):
walk_data(item, pattern, path=f"{path}/{idx}")
yaml = ruamel.yaml.YAML()
data = yaml.load(in_file)
walk_data(data, "changeme")
yaml.dump(data, sys.stdout)
给出:
apiVersion: certmanager.k8s.io/v1alpha1
kind: ClusterIssuer
metadata:
name: letsencrypt-production
spec:
acme:
# You must replace this email address with your own.
# Let's Encrypt will use this to contact you about expiring
# certificates, and issues related to your account.
email: repl_0
server: https://acme-v02.api.letsencrypt.org/directory
privateKeySecretRef:
# Secret resource used to store the account's private key.
name: letsencrypt-production
# Enable the HTTP01 challenge mechanism for this Issuer
dns01:
providers:
- name: prod-cloudflare
cloudflare:
email: repl_1
apiKeySecretRef:
name: cloudflare-api-key-secret
key: 42