在ruamel.yaml中使用自定义构造函数时，如何避免全局状态？

Question

我正在使用ruamel.yaml解析复杂的yaml文档，其中某些标记节点需要特殊处理。我使用add_multi_constructor注入我的自定义解析逻辑，正如已发布的示例所建议的那样。问题是，我需要根据外部状态动态地更改注入的逻辑，但是像add_multi_constructor这样的装饰方法修改了全局状态，从而在逻辑上不相关的实例之间引入了不可接受的耦合。以下是MWE：

import ruamel.yaml

def get_loader(parameter):
    def construct_node(constructor: ruamel.yaml.Constructor, tag: str, node: ruamel.yaml.Node):
        return parameter(tag.lstrip("!"), str(node.value))

    loader = ruamel.yaml.YAML()
    loader.constructor.add_multi_constructor("", construct_node)
    return loader

foo = get_loader(lambda tag, node: f"foo: {tag}, {node}")
bar = get_loader(lambda tag, node: f"bar: {tag}, {node}")
print(foo.load("!abc 123"), bar.load("!xyz 456"), sep="\n")

输出：

bar: abc, 123
bar: xyz, 456

预期：

foo: abc, 123
bar: xyz, 456

我采取了以下变通方法，动态创建新的类实例以打破耦合：

def get_loader(parameter):
    def construct_node(constructor: ruamel.yaml.Constructor, tag: str, node: ruamel.yaml.Node):
        return parameter(tag.lstrip("!"), str(node.value))

    # Create a new class to prevent state sharing through class attributes.
    class ConstructorWrapper(ruamel.yaml.constructor.RoundTripConstructor):
        pass

    loader = ruamel.yaml.YAML()
    loader.Constructor = ConstructorWrapper
    loader.constructor.add_multi_constructor("", construct_node)
    return loader

我的问题是：

• 我是否误用了该库？全局效应是一个巨大的红旗，表明我错误地使用了API，但该库缺少任何API文档，因此我不确定正确的方法是什么

• 从API破损的角度来看是否安全？由于没有记录在案的API，我不确定这是否安全投入生产

Answer 1

在我看来，你没有滥用这个库，只是在解决它目前的缺点/不完整性

在ruamel.yaml获得带有YAML()实例的API之前，它拥有函数 基于PyYAML的API，并带有一些扩展，其他PyYAML的问题必须在 类似的不自然的方式。例如，我恢复使用可以调用其实例的类（使用 __call__()）哪些方法可以更改为只访问 从文档解析的YAML文档版本（因为ruamel.YAML支持YAML 1.2和1.1，仅Pyaml（部分）支持1.1）

但是在ruamel.yaml的YAML()实例下面，并不是所有的情况都有所改善。代码 从PyYAML继承来存储各种构造函数的信息 在类中属性作为查找表（在yaml_constructor上） yaml_multi_constructor），而ruamel.yaml仍然这样做（就像以前一样） PyYAML escque API实际上仍然存在，只有版本0.17才有未来 弃用警告）

到目前为止，您的方法非常有趣，您做到了：

loader.constructor.add_multi_constructor("", construct_node)

而不是：

loader.Constructor.add_multi_constructor("", construct_node)

（您可能知道loader.constructor是一个实例化 loader.Constructor如有必要，但本答案的其他读者可能不会）

甚至：

def get_loader(parameter):
    def construct_node(constructor: ruamel.yaml.Constructor, tag: str, node: ruamel.yaml.Node):
        return parameter(tag.lstrip("!"), str(node.value))

    # Create a new class to prevent state sharing through class attributes.
    class ConstructorWrapper(ruamel.yaml.constructor.RoundTripConstructor):
        pass

    ConstructorWrapper.add_multi_constructor("", construct_node)

    loader = ruamel.yaml.YAML()
    loader.Constructor = ConstructorWrapper
    return loader

代码之所以能够工作，是因为构造函数存储在class属性上，因为.add_multi_constructor()是一个类方法

因此，从API破损的角度来看，您所做的并不是完全安全的。ruamel.yaml不是版本 1.0和（API）可能会破坏代码的更改可能会带来任何问题 小版本号更改。您应该相应地为设置版本依赖项 您的生产代码（例如ruamel.yaml<0.18），并仅在使用新的次要版本号的ruamel.yaml版本进行测试后更新该次要版本号

---

通过更新类属性，可以透明地更改类属性的使用 类方法add_constructor()和add_multi_constructor()为“正常” 方法并在__init__()中完成查找表的初始化。 调用实例的两个示例：

loader.constructor.add_multi_constructor("", construct_node)

将得到预期的结果，但鲁阿梅尔。亚马尔的行为不会改变 使用以下命令对类调用add_multi_constructor时：

loader.Constructor.add_multi_constructor("", construct_node)

但是更改类方法add_constructor()和add_multi_constructor() 以这种方式影响所有代码，这恰好提供了一个实例 而不是类（并表示代码对结果没有影响）

更有可能的是，两个新的 实例方法将被添加到Constructor类和YAML()实例中，并且该类方法将 被逐步淘汰或更改为检查类而不是正在运行的实例 在带有警告的弃用期之后传入（从PyYAML继承的全局函数add_constructor()和add_multi_constructor()）

主要的建议是，除了将生产代码固定在次要代码上之外 版本号，用于确保显示测试代码 PendingDeprecationWarning。如果您使用的是pytest，这就是the case by default。 这将给您足够的时间使代码适应警告的内容 推荐

如果ruamel.yaml的作者不再懒惰，他可能会提供 有关此类API添加/更改的一些文档

import ruamel.yaml
import types
import inspect


class MyConstructor(ruamel.yaml.constructor.RoundTripConstructor):
    _cls_yaml_constructors = {}
    _cls_yaml_multi_constructors = {}

    def __init__(self, *args, **kw):
        self._yaml_constructors = {
            'tag:yaml.org,2002:null': self.__class__.construct_yaml_null,
            'tag:yaml.org,2002:bool': self.__class__.construct_yaml_bool,
            'tag:yaml.org,2002:int': self.__class__.construct_yaml_int,
            'tag:yaml.org,2002:float': self.__class__.construct_yaml_float,
            'tag:yaml.org,2002:binary': self.__class__.construct_yaml_binary,
            'tag:yaml.org,2002:timestamp': self.__class__.construct_yaml_timestamp,
            'tag:yaml.org,2002:omap': self.__class__.construct_yaml_omap,
            'tag:yaml.org,2002:pairs': self.__class__.construct_yaml_pairs,
            'tag:yaml.org,2002:set': self.__class__.construct_yaml_set,
            'tag:yaml.org,2002:str': self.__class__.construct_yaml_str,
            'tag:yaml.org,2002:seq': self.__class__.construct_yaml_seq,
            'tag:yaml.org,2002:map': self.__class__.construct_yaml_map,
            None: self.__class__.construct_undefined
        }
        self._yaml_constructors.update(self._cls_yaml_constructors)
        self._yaml_multi_constructors = self._cls_yaml_multi_constructors.copy()
        super().__init__(*args, **kw)

    def construct_non_recursive_object(self, node, tag=None):
        # type: (Any, Optional[str]) -> Any
        constructor = None  # type: Any
        tag_suffix = None
        if tag is None:
            tag = node.tag
        if tag in self._yaml_constructors:
            constructor = self._yaml_constructors[tag]
        else:
            for tag_prefix in self._yaml_multi_constructors:
                if tag.startswith(tag_prefix):
                    tag_suffix = tag[len(tag_prefix) :]
                    constructor = self._yaml_multi_constructors[tag_prefix]
                    break
            else:
                if None in self._yaml_multi_constructors:
                    tag_suffix = tag
                    constructor = self._yaml_multi_constructors[None]
                elif None in self._yaml_constructors:
                    constructor = self._yaml_constructors[None]
                elif isinstance(node, ScalarNode):
                    constructor = self.__class__.construct_scalar
                elif isinstance(node, SequenceNode):
                    constructor = self.__class__.construct_sequence
                elif isinstance(node, MappingNode):
                    constructor = self.__class__.construct_mapping
        if tag_suffix is None:
            data = constructor(self, node)
        else:
            data = constructor(self, tag_suffix, node)
        if isinstance(data, types.GeneratorType):
            generator = data
            data = next(generator)
            if self.deep_construct:
                for _dummy in generator:
                    pass
            else:
                self.state_generators.append(generator)
        return data

    def get_args(*args, **kw):
        if kw:
            raise NotImplementedError('can currently only handle positional arguments')
        if len(args) == 2:
            return MyConstructor, args[0], args[1]
        else:
            return args[0], args[1], args[2]

    def add_constructor(self, tag, constructor):
        self, tag, constructor = MyConstructor.get_args(*args, **kw)
        if inspect.isclass(self):
            self._cls_yaml_constructors[tag] = constructor
            return
        self._yaml_constructors[tag] = constructor

    def add_multi_constructor(*args, **kw): # self, tag_prefix, multi_constructor):
        self, tag_prefix, multi_constructor = MyConstructor.get_args(*args, **kw)
        if inspect.isclass(self):
            self._cls_yaml_multi_constructors[tag_prefix] = multi_constructor
            return
        self._yaml_multi_constructors[tag_prefix] = multi_constructor

def get_loader_org(parameter):
    def construct_node(constructor: ruamel.yaml.Constructor, tag: str, node: ruamel.yaml.Node):
        return parameter(tag.lstrip("!"), str(node.value))

    loader = ruamel.yaml.YAML()
    loader.Constructor = MyConstructor
    loader.constructor.add_multi_constructor("", construct_node)
    return loader

foo = get_loader_org(lambda tag, node: f"foo: {tag}, {node}")
bar = get_loader_org(lambda tag, node: f"bar: {tag}, {node}")
print('>org<', foo.load("!abc 123"), bar.load("!xyz 456"), sep="\n")


def get_loader_instance(parameter):
    def construct_node(constructor: ruamel.yaml.Constructor, tag: str, node: ruamel.yaml.Node):
        return parameter(tag.lstrip("!"), str(node.value))

    # Create a new class to prevent state sharing through class attributes.
    class ConstructorWrapper(MyConstructor):
        pass

    loader = ruamel.yaml.YAML()
    loader.Constructor = ConstructorWrapper
    loader.constructor.add_multi_constructor("", construct_node)
    return loader

foo = get_loader_instance(lambda tag, node: f"foo: {tag}, {node}")
bar = get_loader_instance(lambda tag, node: f"bar: {tag}, {node}")
print('>instance<', foo.load("!abc 123"), bar.load("!xyz 456"), sep="\n")


def get_loader_cls(parameter):
    def construct_node(constructor: ruamel.yaml.Constructor, tag: str, node: ruamel.yaml.Node):
        return parameter(tag.lstrip("!"), str(node.value))

    # Create a new class to prevent state sharing through class attributes.
    class ConstructorWrapper(MyConstructor):
        pass

    loader = ruamel.yaml.YAML()
    loader.Constructor = ConstructorWrapper
    loader.Constructor.add_multi_constructor("", construct_node)
    #      ^ using the virtual class method
    return loader

foo = get_loader_cls(lambda tag, node: f"foo: {tag}, {node}")
bar = get_loader_cls(lambda tag, node: f"bar: {tag}, {node}")
print('>cls<', foo.load("!abc 123"), bar.load("!xyz 456"), sep="\n")

其中：

>org<
foo: abc, 123
bar: xyz, 456
>instance<
foo: abc, 123
bar: xyz, 456
>cls<
bar: abc, 123
bar: xyz, 456