解析嵌套的自定义 yaml 标签

Question

我有一些带有应用程序特定标签的 yaml（确切地说，来自 AWS Cloud Formation 模板），如下所示：

example_yaml = "Name: !Join [' ', ['EMR', !Ref 'Environment', !Ref 'Purpose']]"

我想解析它以便我可以这样做:

>>> print(result)
>>> {'Name': 'EMR {Environment} {Purpose}'}

>>> name = result['name'].format(
...    Environment='Development',
...    Purpose='ETL'
... )
>>> print(name)
>>> EMR Development ETL

目前我的代码如下所示：

import yaml
from pprint import pprint


def aws_join(loader, node):
    join_args = loader.construct_yaml_seq(node)
    delimiter = list(join_args)[0]
    joinables = list(join_args)[1]
    join_result = delimiter.join(joinables)
    return join_result

def aws_ref(loader, node):
    value = loader.construct_scalar(node)
    placeholder = '{'+value+'}'
    return placeholder

yaml.add_constructor('!Join', aws_join)
yaml.add_constructor('!Ref', aws_ref)

example_yaml = "Name: !Join [' ', ['EMR', !Ref 'Environment', !Ref 'Purpose']]"

pprint(yaml.load(example_yaml))

不幸的是，这会导致错误。

...
   joinables = list(join_args)[1]
IndexError: list index out of range

将 print('What I am: '+str(join_args)) 添加到 aws_join 表明我得到了一个生成器：

What I am: <generator object SafeConstructor.construct_yaml_seq at 0x1082ece08>

这就是我尝试将生成器转换为列表的原因。不过，生成器最终会正确填充，只是没有及时让我使用它。如果我将 aws_join 函数更改为这样：

def aws_join(loader, node):
    join_args = loader.construct_yaml_seq(node)
    return join_args

那么最后的结果是这样的：

{'Name': [' ', ['EMR', '{Environment}', '{Purpose}']]}

所以我的函数所需的部分都在那里，只是当我在我的函数中需要它们时没有。

Answer 1

您需要更改：

def aws_join(loader, node):
    delimiter = loader.construct_scalar(node.value[0])
    value = loader.construct_sequence(node.value[1])
    return delimiter.join(value)

然后你会得到输出：

{'Name': 'EMR {Environment} {Purpose}'}

Answer 2

你很接近，但问题是你使用的方法 construct_yaml_seq()。该方法实际上是注册的普通 YAML 序列的构造函数（最终使 Python 列表）并调用 construct_sequence() 方法来处理传入的节点，这也是您应该做的。

因为你返回的是字符串，无法处理递归数据结构，您不需要使用两步创建过程（首先 yield-ing，然后填写）其中construct_yaml_seq()方法如下。但这两个步骤的创建过程是 为什么你遇到了发电机.

construct_sequence returns 一个简单的列表，但是你想要的节点在开始处理时可用的 !Join 下面，确保指定 deep=True 参数，否则第二个列表元素将是一个空列表。而且因为 construct_yaml_seq(), 没有指定deep=True，你没有及时拿到碎片你的函数（否则你实际上可以使用那个方法）。

import yaml
from pprint import pprint


def aws_join(loader, node):
    join_args = loader.construct_sequence(node, deep=True)
    # you can comment out next line
    assert join_args == [' ', ['EMR', '{Environment}', '{Purpose}']] 
    delimiter = join_args[0]
    joinables = join_args[1]
    return delimiter.join(joinables)

def aws_ref(loader, node):
    value = loader.construct_scalar(node)
    placeholder = '{'+value+'}'
    return placeholder

yaml.add_constructor('!Join', aws_join, Loader=yaml.SafeLoader)
yaml.add_constructor('!Ref', aws_ref, Loader=yaml.SafeLoader)

example_yaml = "Name: !Join [' ', ['EMR', !Ref 'Environment', !Ref 'Purpose']]"

pprint(yaml.safe_load(example_yaml))

给出：

{'Name': 'EMR {Environment} {Purpose}'}

你不应该使用 load()，它被记录为潜在的不安全，最重要的是：这里没有必要。注册 SafeLoader 并调用 safe_load()

解析嵌套的自定义 yaml 标签

Parse nested custom yaml tags

python

yaml

pyyaml