Obteniendo claves duplicadas en YAML usando Python

Necesitamos analizar archivos YAML que contengan claves duplicadas y todos estos deben ser analizados. No basta con omitir duplicados. Sé que esto va en contra de la especificación de YAML y me gustaría no tener que hacerlo, pero una herramienta de terceros que usamos permite este uso y debemos lidiar con eso.

Ejemplo de archivo:

build: step: 'step1' build: step: 'step2' 

Después de analizar debemos tener una estructura de datos similar a esta:

 yaml.load('file.yml') # [('build', [('step', 'step1')]), ('build', [('step', 'step2')])] 

dict ya no se puede utilizar para representar los contenidos analizados.

Estoy buscando una solución en Python y no encontré una biblioteca que lo respalde, ¿me he perdido algo?

Alternativamente, estoy feliz de escribir mi propia cosa, pero me gustaría hacerlo lo más simple posible. ruamel.yaml parece el analizador de YAML más avanzado en Python y parece moderadamente extensible, ¿se puede extender para admitir campos duplicados?

PyYAML solo sobrescribirá silenciosamente la primera entrada, ruamel.yaml give le dará a DuplicateKeyFutureWarning si se usa con la API heredada, y generará una DuplicateKeyError con la nueva API.

Si no desea crear un Constructor completo para todos los tipos, sobrescribir el constructor de mapeo en SafeConstructor debe hacer el trabajo:

 import sys from ruamel.yaml import YAML from ruamel.yaml.constructor import SafeConstructor yaml_str = """\ build: step: 'step1' build: step: 'step2' """ def construct_yaml_map(self, node): # test if there are duplicate node keys data = [] yield data for key_node, value_node in node.value: key = self.construct_object(key_node, deep=True) val = self.construct_object(value_node, deep=True) data.append((key, val)) SafeConstructor.add_constructor(u'tag:yaml.org,2002:map', construct_yaml_map) yaml = YAML(typ='safe') data = yaml.load(yaml_str) print(data) 

lo que da:

 [('build', [('step', 'step1')]), ('build', [('step', 'step2')])] 

Sin embargo, no parece necesario hacer el step: 'step1' en una lista. Lo siguiente solo creará la lista si hay elementos duplicados (podría optimizarse si es necesario, almacenando en caché el resultado del self.construct_object(key_node, deep=True) ):

 def construct_yaml_map(self, node): # test if there are duplicate node keys keys = set() for key_node, value_node in node.value: key = self.construct_object(key_node, deep=True) if key in keys: break keys.add(key) else: data = {} # type: Dict[Any, Any] yield data value = self.construct_mapping(node) data.update(value) return data = [] yield data for key_node, value_node in node.value: key = self.construct_object(key_node, deep=True) val = self.construct_object(value_node, deep=True) data.append((key, val)) 

lo que da:

 [('build', {'step': 'step1'}), ('build', {'step': 'step2'})] 

Algunos puntos:

  • Probablemente no hace falta decir que esto no funcionará con las teclas de combinación YAML ( <<: *xyz )
  • Si necesita las capacidades de ida y vuelta de yaml = YAML() ( yaml = YAML() ), eso requerirá un construct_yaml_map más complejo.
  • Si desea volcar la salida, debe crear una instancia de una nueva instancia de YAML() para eso, en lugar de reutilizar la "parcheada" utilizada para cargar (puede que funcione, esto es solo para estar seguro):

     yaml_out = YAML(typ='safe') yaml_out.dump(data, sys.stdout) 

    que da (con el primer construct_yaml_map ):

     - - build - - [step, step1] - - build - - [step, step2] 
  • Lo que no funciona en PyYAML ni en ruamel.yaml es yaml.load('file.yml') . Si no quieres open() el archivo, puedes hacerlo:

     from pathlib import Path # or: from ruamel.std.pathlib import Path yaml = YAML(typ='safe') yaml.load(Path('file.yml') 

¹ Descargo de responsabilidad: Soy el autor de ese paquete.

Si puede modificar los datos de entrada muy ligeramente, debería poder hacer esto convirtiendo el único archivo tipo yaml en varios documentos yaml. Los documentos yaml pueden estar en el mismo archivo si están separados por --- en una línea por sí mismo, y parece que tiene entradas separadas por dos nuevas líneas una al lado de la otra:

 with open('file.yml', 'r') as f: data = f.read() data = data.replace('\n\n', '\n---\n') for document in yaml.load_all(data): print(document) 

Salida:

 {'build': {'step': 'step1'}} {'build': {'step': 'step2'}} 

Puede anular cómo Pyyaml ​​carga las claves. Por ejemplo, podría usar un código predeterminado con listas de valores para cada clave:

 from collections import defaultdict import yaml def parse_preserving_duplicates(src): # We deliberately define a fresh class inside the function, # because add_constructor is a class method and we don't want to # mutate pyyaml classes. class PreserveDuplicatesLoader(yaml.loader.Loader): pass def map_constructor(loader, node, deep=False): """Walk the mapping, recording any duplicate keys. """ mapping = defaultdict(list) for key_node, value_node in node.value: key = loader.construct_object(key_node, deep=deep) value = loader.construct_object(value_node, deep=deep) mapping[key].append(value) return mapping PreserveDuplicatesLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, map_constructor) return yaml.load(src, PreserveDuplicatesLoader)