python - 如何将 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() 不是这些，所以上面的例子不起作用。

实际输出:

$ 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，您可以编写自己的类来执行您想要的操作:

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 类为您提供了一个带有 read 方法的对象，该方法会愉快地遍历文件列表，直到所有内容都用完。鉴于上述代码和您的样本输入，我们得到了您正在寻找的输出。