在 ruamel.yaml 中使用自定义构造函数时如何避免全局状态？
How do I avoid global state when using custom constructors in ruamel.yaml?
我正在使用 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
我做了以下解决方法，动态创建新的 class 实例以打破耦合：
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，我不是确定这是否可以安全投入生产。
IMO 你没有滥用这个库，只是在解决它当前的问题 shortcomings/incompleteness。
在ruamel.yaml用YAML()实例得到API之前，它有函数
基于 API 的 PyYAML，带有一些扩展，其他 PyYAML 的问题必须在
类似不自然的方式。例如。我恢复到可以调用 classes 的实例（使用
__call__()) 然后可以将哪些方法更改为只能访问
从文档解析的 YAML 文档版本（因为 ruamel.yaml 支持 YAML
1.2 和 1.1 以及 PyYAML 仅（部分）支持 1.1）。
但在 ruamel.yaml 的 YAML() 下，并非所有情况都得到改善。代码
继承自 PyYAML 存储各种构造函数的信息
在 class 属性中作为查找表（在 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() 是一个 class 方法。
因此，就 API 破损而言，您所做的并不完全安全。 ruamel.yaml 不是版本
1.0，并且 (API) 可能会破坏您的代码的更改可能会随任何
次要版本号更改。你应该相应地设置你的版本依赖
您的生产代码（例如 ruamel.yaml<0.18 ），并仅在使用具有新次要版本号的 ruamel.yaml 版本进行测试后更新次要编号。
可以通过更新透明地更改 class 属性的使用
class 方法 add_constructor() 和 add_multi_constructor() 到“正常”
方法并在 __init__() 中完成查找表的初始化。
您调用实例的两个示例：
loader.constructor.add_multi_constructor("", construct_node)
会得到想要的结果，但是ruamel.yaml的行为不会改变
在 class 上调用 add_multi_constructor 时使用：
loader.Constructor.add_multi_constructor("", construct_node)
但是改变 class 方法 add_constructor() 和 add_multi_constructor()
以这种方式影响所有代码，这恰好提供了一个实例
而不是 class（并且说代码对结果没问题）。
更有可能是两个新的
实例方法将添加到 Constructor class 和 YAML() 实例，并且 class 方法将
被淘汰或更改为检查 class 而不是正在检查的实例
在带有警告的弃用期后传入（从 PyYAML 继承的全局函数 add_constructor() 和 add_multi_constructor() 也是如此）。
主要建议，除了将您的生产代码固定在次要建议上
版本号，是为了确保你的测试代码显示
PendingDeprecationWarning。如果您使用 pytest 这是 的情况
默认。
这应该给你足够的时间来调整你的代码以适应警告
推荐。
如果 ruamel.yaml 的作者不再懒惰，他可能会提供
API additions/changes.
的一些文档
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
我正在使用 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
我做了以下解决方法，动态创建新的 class 实例以打破耦合：
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，我不是确定这是否可以安全投入生产。
IMO 你没有滥用这个库，只是在解决它当前的问题 shortcomings/incompleteness。
在ruamel.yaml用YAML()实例得到API之前，它有函数
基于 API 的 PyYAML，带有一些扩展，其他 PyYAML 的问题必须在
类似不自然的方式。例如。我恢复到可以调用 classes 的实例（使用
__call__()) 然后可以将哪些方法更改为只能访问
从文档解析的 YAML 文档版本（因为 ruamel.yaml 支持 YAML
1.2 和 1.1 以及 PyYAML 仅（部分）支持 1.1）。
但在 ruamel.yaml 的 YAML() 下，并非所有情况都得到改善。代码
继承自 PyYAML 存储各种构造函数的信息
在 class 属性中作为查找表（在 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() 是一个 class 方法。
因此，就 API 破损而言，您所做的并不完全安全。 ruamel.yaml 不是版本
1.0，并且 (API) 可能会破坏您的代码的更改可能会随任何
次要版本号更改。你应该相应地设置你的版本依赖
您的生产代码（例如 ruamel.yaml<0.18 ），并仅在使用具有新次要版本号的 ruamel.yaml 版本进行测试后更新次要编号。
可以通过更新透明地更改 class 属性的使用
class 方法 add_constructor() 和 add_multi_constructor() 到“正常”
方法并在 __init__() 中完成查找表的初始化。
您调用实例的两个示例：
loader.constructor.add_multi_constructor("", construct_node)
会得到想要的结果，但是ruamel.yaml的行为不会改变
在 class 上调用 add_multi_constructor 时使用：
loader.Constructor.add_multi_constructor("", construct_node)
但是改变 class 方法 add_constructor() 和 add_multi_constructor()
以这种方式影响所有代码，这恰好提供了一个实例
而不是 class（并且说代码对结果没问题）。
更有可能是两个新的
实例方法将添加到 Constructor class 和 YAML() 实例，并且 class 方法将
被淘汰或更改为检查 class 而不是正在检查的实例
在带有警告的弃用期后传入（从 PyYAML 继承的全局函数 add_constructor() 和 add_multi_constructor() 也是如此）。
主要建议，除了将您的生产代码固定在次要建议上
版本号，是为了确保你的测试代码显示
PendingDeprecationWarning。如果您使用 pytest 这是 的情况
默认。
这应该给你足够的时间来调整你的代码以适应警告
推荐。
如果 ruamel.yaml 的作者不再懒惰，他可能会提供 API additions/changes.
的一些文档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