如何使用 yaml.load_all 和 fileinput.input？

Question

在不求助于 ''.join 的情况下，是否有一种 Pythonic 方式使用 PyYAML 的 yaml.load_all 和 fileinput.input() 来轻松流式传输来自多个来源的多个文档？

我正在寻找类似下面的内容（非工作示例）：

# example.py
import fileinput

import yaml

for doc in yaml.load_all(fileinput.input()):
    print(doc)

预期输出：

$ cat >pre.yaml <<<'--- prefix-doc'
$ cat >post.yaml <<<'--- postfix-doc'
$ python example.py pre.yaml - post.yaml <<<'--- hello'
prefix-doc
hello
postfix-doc

当然，yaml.load_all 需要字符串、字节或类似文件的对象，而 fileinput.input() 是这些东西的 none，因此上面的示例不起作用。

实际输出：

$ python example.py pre.yaml - post.yaml <<<'--- hello'
...
AttributeError: FileInput instance has no attribute 'read'

您可以使示例与 ''.join 一起工作，但那是作弊。我正在寻找一种不会立即将整个流读入内存的方法。

我们可以将问题改写为 是否有某种方法可以模拟字符串、字节或类似文件的对象来代理字符串的底层迭代器？ 但是，我怀疑yaml.load_all 实际上需要整个类似文件的界面，因此措辞会要求比绝对必要的更多。

理想情况下，我正在寻找能够支持如下内容的最小适配器：

for doc in yaml.load_all(minimal_adapter(fileinput.input())):
    print(doc)

Answer 1

fileinput.input 的问题在于生成的对象没有 read 方法，而这正是 yaml.load_all 所寻找的。如果你愿意放弃 fileinput，你可以自己编写 class 来做你想做的事情：

import sys                                                                      
import yaml                                                                     

class BunchOFiles (object):                                                     
    def __init__(self, *files):                                                 
        self.files = files                                                      
        self.fditer = self._fditer()                                            
        self.fd = self.fditer.next()                                            

    def _fditer(self):                                                          
        for fn in self.files:                                                   
            with sys.stdin if fn == '-' else open(fn, 'r') as fd:               
                yield fd                                                        

    def read(self, size=-1):                                                    
        while True:                                                             
            data = self.fd.read(size)                                           

            if data:                                                            
                break                                                           
            else:                                                               
                try:                                                            
                    self.fd = self.fditer.next()                                
                except StopIteration:                                           
                    self.fd = None                                              
                    break                                                       

        return data                                                             

bunch = BunchOFiles(*sys.argv[1:])                                              
for doc in yaml.load_all(bunch):                                                
    print doc

BunchOFiles class 为您提供了一个具有 read 方法的对象，该方法将愉快地遍历文件列表，直到所有内容都用完。鉴于上述代码和您的样本输入，我们得到了您正在寻找的输出。

Answer 2

您的 minimal_adapter 应该将 fileinput.FileInput 作为参数，return 是 load_all 可以使用的对象。 load_all 将字符串作为参数，但这需要连接输入，或者它期望参数具有 read() 方法。

由于您的 minimal_adapter 需要保留一些状态，我发现 clearest/easiest 将其实现为具有 __call__ 方法的 class 的实例，并且拥有该方法 return 实例并存储其参数以供将来使用。以这种方式实现，class 也应该有一个 read() 方法，因为这将在将实例交给 load_all:

之后调用

import fileinput
import ruamel.yaml


class MinimalAdapter:
    def __init__(self):
        self._fip = None
        self._buf = None  # storage of read but unused material, maximum one line

    def __call__(self, fip):
        self._fip = fip  # store for future use
        self._buf = ""
        return self

    def read(self, size):
        if len(self._buf) >= size:
            # enough in buffer from last read, just cut it off and return
            tmp, self._buf = self._buf[:size], self._buf[size:]
            return tmp
        for line in self._fip:
            self._buf += line
            if len(self._buf) > size:
                break
        else:
            # ran out of lines, return what we have
            tmp, self._buf = self._buf, ''
            return tmp
        tmp, self._buf = self._buf[:size], self._buf[size:]
        return tmp


minimal_adapter = MinimalAdapter()

for doc in ruamel.yaml.load_all(minimal_adapter(fileinput.input())):
    print(doc)

这样，运行您的示例调用就可以准确地给出您想要的输出。

对于较大的文件，这可能只会提高内存效率。 load_all 尝试一次读取 1024 字节块（通过在 MinimalAdapter.read() 中放置打印语句很容易发现）并且 fileinput 也做一些缓冲（使用 strace 如果您有兴趣了解它的行为方式）。

_{这是使用 ruamel.yaml YAML 1.2 解析器完成的，我是其中的作者。这应该适用于 PyYAML，其中 ruamel.yaml 也是派生超集。}

如何使用 yaml.load_all 和 fileinput.input？

How to use yaml.load_all with fileinput.input?

python

pyyaml