Разбор элементов списка yaml по одному в python

Question

Есть ли в Python библиотека yaml, которая может при необходимости читать входной файл за раз, а не анализировать весь файл? У меня есть длинный файл со списком в качестве корневого узла. Если я пытаюсь найти первый элемент, удовлетворяющий определенному свойству, мне может не понадобиться читать и анализировать весь файл и быстрее получать результат.

Answer 1

Вы можете использовать низкоуровневый API PyYAML parse():

import yaml

for event in yaml.parse(input):
    # process event

События задокументированы здесь .

Если вы хотите создать каждый элементпоследовательность корневого уровня в собственное значение Python, вам нужно использовать классы Composer и Constructor. Composer читает события и преобразует их в узлы , Constructor строит значения Python из узлов . Это соответствует процессу загрузки, определенному в спецификации YAML:

Теперь PyYAML Composer ожидает, что функции get_event, check_event и peek_event будут существовать в self, но не реализует их. Они реализованы Parser. Поэтому, чтобы иметь работающую цепочку загрузки YAML, PyYAML позже делает:

class Loader(Reader, Scanner, Parser, Composer, Constructor, Resolver):
  def __init__(self, stream):
    Reader.__init__(self, stream)
    Scanner.__init__(self)
    Parser.__init__(self)
    Composer.__init__(self)
    Constructor.__init__(self)
    Resolver.__init__(self)

Для вас это означает, что вам нужен объект Loader и использование API Parser для событий верхнего уровня, а такжес помощью Composer и Constructor API для загрузки каждого элемента в последовательности верхнего уровня.

Вот некоторый код, с которого можно начать:

import yaml

input = """
- "A": 1
- "B": 2
- foo
- 1
"""

loader = yaml.SafeLoader(input)

# check proper stream start (should never fail)
assert loader.check_event(yaml.StreamStartEvent)
loader.get_event()
assert loader.check_event(yaml.DocumentStartEvent)
loader.get_event()

# assume the root element is a sequence
assert loader.check_event(yaml.SequenceStartEvent)
loader.get_event()

# now while the next event does not end the sequence, process each item
while not loader.check_event(yaml.SequenceEndEvent):
    # compose current item to a node as if it was the root node
    node = loader.compose_node(None, None)
    # construct a native Python value with the node.
    # we set deep=True for complete processing of all the node's children
    value = loader.construct_object(node, True)
    print(value)

# assume document ends and no further documents are in stream
loader.get_event()
assert loader.check_event(yaml.DocumentEndEvent)
loader.get_event()
assert loader.check_event(yaml.StreamEndEvent)

Помните, что вы можете запуститьЕсли у вас есть якоря и псевдонимы в документе YAML.