我有两个YAML文件,“A”和“B”,我希望将A的内容插入到B中,要么拼接到现有的数据结构中,就像数组一样,要么作为元素的子元素,就像某个散列键的值一样。

这可能吗?怎么做?如果不是,是否有指向规范引用的指针?


当前回答

结合其他答案,这里是一个简短的解决方案,没有重载Loader类,它可以与任何加载器操作文件:

import json
from pathlib import Path
from typing import Any

import yaml


def yaml_include_constructor(loader: yaml.BaseLoader, node: yaml.Node) -> Any:
    """Include file referenced with !include node"""

    # noinspection PyTypeChecker
    fp = Path(loader.name).parent.joinpath(loader.construct_scalar(node)).resolve()
    fe = fp.suffix.lstrip(".")

    with open(fp, 'r') as f:
        if fe in ("yaml", "yml"):
            return yaml.load(f, type(loader))
        elif fe in ("json", "jsn"):
            return json.load(f)
        else:
            return f.read()


def main():
    loader = yaml.SafeLoader  # Works with any loader
    loader.add_constructor("!include", yaml_include_constructor)

    with open(...) as f:
        yml = yaml.load(f, loader)

PyTypeChecker的存在是为了防止pep检查警告预期类型'ScalarNode',得到'节点'而不是通过节点:yaml。节点到loader.construct_scalar()。

如果yaml。加载输入流不是文件流,因为loader.name在这种情况下不包含路径:

class Reader(object):
    ...
    def __init__(self, stream):
        ...
        if isinstance(stream, str):
            self.name = "<unicode string>"
            ...
        elif isinstance(stream, bytes):
            self.name = "<byte string>"
            ...
        else:
            self.name = getattr(stream, 'name', "<file>")
            ...

在我的用例中,我知道只包含YAML文件,所以解决方案可以进一步简化:

def yaml_include_constructor(loader: yaml.Loader, node: yaml.Node) -> Any:
    """Include YAML file referenced with !include node"""
    with open(Path(loader.name).parent.joinpath(loader.construct_yaml_str(node)).resolve(), 'r') as f:
        return yaml.load(f, type(loader))


Loader = yaml.SafeLoader  # Works with any loader
Loader.add_constructor("!include", yaml_include_constructor)


def main():
    with open(...) as f:
        yml = yaml.load(f, Loader=Loader)

甚至是使用lambda的一行代码:

Loader = yaml.SafeLoader  # Works with any loader
Loader.add_constructor("!include",
                       lambda l, n: yaml.load(Path(l.name).parent.joinpath(l.construct_scalar(n)).read_text(), type(l)))

其他回答

我举了一些例子供你参考。

import yaml

main_yaml = """
Package:
 - !include _shape_yaml    
 - !include _path_yaml
"""

_shape_yaml = """
# Define
Rectangle: &id_Rectangle
    name: Rectangle
    width: &Rectangle_width 20
    height: &Rectangle_height 10
    area: !product [*Rectangle_width, *Rectangle_height]

Circle: &id_Circle
    name: Circle
    radius: &Circle_radius 5
    area: !product [*Circle_radius, *Circle_radius, pi]

# Setting
Shape:
    property: *id_Rectangle
    color: red
"""

_path_yaml = """
# Define
Root: &BASE /path/src/

Paths: 
    a: &id_path_a !join [*BASE, a]
    b: &id_path_b !join [*BASE, b]

# Setting
Path:
    input_file: *id_path_a
"""


# define custom tag handler
def yaml_import(loader, node):
    other_yaml_file = loader.construct_scalar(node)
    return yaml.load(eval(other_yaml_file), Loader=yaml.SafeLoader)


def yaml_product(loader, node):
    import math
    list_data = loader.construct_sequence(node)
    result = 1
    pi = math.pi
    for val in list_data:
        result *= eval(val) if isinstance(val, str) else val
    return result


def yaml_join(loader, node):
    seq = loader.construct_sequence(node)
    return ''.join([str(i) for i in seq])


def yaml_ref(loader, node):
    ref = loader.construct_sequence(node)
    return ref[0]


def yaml_dict_ref(loader: yaml.loader.SafeLoader, node):
    dict_data, key, const_value = loader.construct_sequence(node)
    return dict_data[key] + str(const_value)


def main():
    # register the tag handler
    yaml.SafeLoader.add_constructor(tag='!include', constructor=yaml_import)
    yaml.SafeLoader.add_constructor(tag='!product', constructor=yaml_product)
    yaml.SafeLoader.add_constructor(tag='!join', constructor=yaml_join)
    yaml.SafeLoader.add_constructor(tag='!ref', constructor=yaml_ref)
    yaml.SafeLoader.add_constructor(tag='!dict_ref', constructor=yaml_dict_ref)

    config = yaml.load(main_yaml, Loader=yaml.SafeLoader)

    pk_shape, pk_path = config['Package']
    pk_shape, pk_path = pk_shape['Shape'], pk_path['Path']
    print(f"shape name: {pk_shape['property']['name']}")
    print(f"shape area: {pk_shape['property']['area']}")
    print(f"shape color: {pk_shape['color']}")

    print(f"input file: {pk_path['input_file']}")


if __name__ == '__main__':
    main()

输出

shape name: Rectangle
shape area: 200
shape color: red
input file: /path/src/a

更新2

你可以把它们结合起来,像这样

# xxx.yaml
CREATE_FONT_PICTURE:
  PROJECTS:
    SUNG: &id_SUNG
      name: SUNG
      work_dir: SUNG
      output_dir: temp
      font_pixel: 24


  DEFINE: &id_define !ref [*id_SUNG]  # you can use config['CREATE_FONT_PICTURE']['DEFINE'][name, work_dir, ... font_pixel]
  AUTO_INIT:
    basename_suffix: !dict_ref [*id_define, name, !product [5, 3, 2]]  # SUNG30

# ↓ This is not correct.
# basename_suffix: !dict_ref [*id_define, name, !product [5, 3, 2]]  # It will build by Deep-level. id_define is Deep-level: 2. So you must put it after 2. otherwise, it can't refer to the correct value.

标准YAML 1.2本身不包括这个特性。尽管如此,许多实现提供了一些扩展来实现这一点。

我给出了一种使用Java和snakeyaml:1.24(用于解析/发出YAML文件的Java库)来实现它的方法,它允许创建一个自定义YAML标记来实现以下目标(你会看到我用它来加载在几个YAML文件中定义的测试套件,并且我让它作为一个目标测试:节点的包含列表工作):

# ... yaml prev stuff

tests: !include
  - '1.hello-test-suite.yaml'
  - '3.foo-test-suite.yaml'
  - '2.bar-test-suite.yaml'

# ... more yaml document

下面是允许处理!include标记的单类Java。文件从classpath (Maven资源目录)加载:

/**
 * Custom YAML loader. It adds support to the custom !include tag which allows splitting a YAML file across several
 * files for a better organization of YAML tests.
 */
@Slf4j   // <-- This is a Lombok annotation to auto-generate logger
public class MyYamlLoader {

    private static final Constructor CUSTOM_CONSTRUCTOR = new MyYamlConstructor();

    private MyYamlLoader() {
    }

    /**
     * Parse the only YAML document in a stream and produce the Java Map. It provides support for the custom !include
     * YAML tag to split YAML contents across several files.
     */
    public static Map<String, Object> load(InputStream inputStream) {
        return new Yaml(CUSTOM_CONSTRUCTOR)
                .load(inputStream);
    }


    /**
     * Custom SnakeYAML constructor that registers custom tags.
     */
    private static class MyYamlConstructor extends Constructor {

        private static final String TAG_INCLUDE = "!include";

        MyYamlConstructor() {
            // Register custom tags
            yamlConstructors.put(new Tag(TAG_INCLUDE), new IncludeConstruct());
        }

        /**
         * The actual include tag construct.
         */
        private static class IncludeConstruct implements Construct {

            @Override
            public Object construct(Node node) {
                List<Node> inclusions = castToSequenceNode(node);
                return parseInclusions(inclusions);
            }

            @Override
            public void construct2ndStep(Node node, Object object) {
                // do nothing
            }

            private List<Node> castToSequenceNode(Node node) {
                try {
                    return ((SequenceNode) node).getValue();

                } catch (ClassCastException e) {
                    throw new IllegalArgumentException(String.format("The !import value must be a sequence node, but " +
                            "'%s' found.", node));
                }
            }

            private Object parseInclusions(List<Node> inclusions) {

                List<InputStream> inputStreams = inputStreams(inclusions);

                try (final SequenceInputStream sequencedInputStream =
                             new SequenceInputStream(Collections.enumeration(inputStreams))) {

                    return new Yaml(CUSTOM_CONSTRUCTOR)
                            .load(sequencedInputStream);

                } catch (IOException e) {
                    log.error("Error closing the stream.", e);
                    return null;
                }
            }

            private List<InputStream> inputStreams(List<Node> scalarNodes) {
                return scalarNodes.stream()
                        .map(this::inputStream)
                        .collect(toList());
            }

            private InputStream inputStream(Node scalarNode) {
                String filePath = castToScalarNode(scalarNode).getValue();
                final InputStream is = getClass().getClassLoader().getResourceAsStream(filePath);
                Assert.notNull(is, String.format("Resource file %s not found.", filePath));
                return is;
            }

            private ScalarNode castToScalarNode(Node scalarNode) {
                try {
                    return ((ScalarNode) scalarNode);

                } catch (ClassCastException e) {
                    throw new IllegalArgumentException(String.format("The value must be a scalar node, but '%s' found" +
                            ".", scalarNode));
                }
            }
        }

    }

}

使用Symfony,它对yaml的处理将间接地允许您嵌套yaml文件。诀窍在于使用参数选项。例如:

common.yml

parameters:
    yaml_to_repeat:
        option: "value"
        foo:
            - "bar"
            - "baz"

config.yml

imports:
    - { resource: common.yml }
whatever:
    thing: "%yaml_to_repeat%"
    other_thing: "%yaml_to_repeat%"

其结果将与:

whatever:
    thing:
        option: "value"
        foo:
            - "bar"
            - "baz"
    other_thing:
        option: "value"
        foo:
            - "bar"
            - "baz"

可能在问问题时不支持,但你可以将其他YAML文件导入其中:

imports: [/your_location_to_yaml_file/Util.area.yaml]

虽然我没有任何在线参考资料,但这对我来说很有用。

根据之前的帖子:

  class SimYamlLoader(yaml.SafeLoader):
        '''
        Simple custom yaml loader that supports include, e.g:

        main.yaml:

        - !include file1.yaml
        - !include dir/file2.yaml

        '''

        def __init__(self, stream):
            self.root = os.path.split(stream.name)[0]
            super().__init__(stream)

    def _include(loader, node):
        filename = os.path.join(loader.root, loader.construct_scalar(node))
        with open(filename, 'r') as f:
            return yaml.load(f, SimYamlLoader)
    SimYamlLoader.add_constructor('!include', _include)

    # example:
    with open('main.yaml', 'r') as f:
        lists = yaml.load(f, SimYamlLoader)
        # if you want to merge the lists
        data = functools.reduce(
            lambda x, y: x if y is None else {**x, **dict(y)}, lists, {})
        # python 3.10+:lambda x, y: x if y is None else x | dict(y), lists, {})