使用python io从缓冲流组成行读取器

Question

我使用python boto与s3交互。我在s3上的文件是CSV的，我想从s3中读取行，使用缓冲区来绑定内存使用。

我想知道是否有人有任何方法可以编写python的io类来实现这一点？我们的目标是有某种抽象，能够包装boto键，并在键上提供一个readline或迭代器接口（只提供read（size=0）调用）。其复杂性在于，由于它存储为CSV，所以每一行的长度都是可变的。

目标是有一个抽象，我能够包装python boto key，然后实现迭代器协议，这样我就可以把它传递给csv阅读器，我最终实现了自己。

它看起来像pythonio真的有所有的部分来完成这个BufferedReader和TextIOWrapper，我天真地试图将botoKey传递给它，但是BufferedReader需要一个IOBase对象。

然后，我围绕密钥实现了IOBase协议，但得到了Unicode错误，只是一般不确定我在做什么。

有人知道python io是否可以做类似于上面描述的事情吗？？

技术规格：

s3上有一个1-100个CSV文件的目录。它们都具有相同的格式，但行数可变。我试图实现一个函数，它需要一个botoKeys的迭代器。

Key提供了一个读取（num_bytes）方法。

def yield_lines(keys_iterator):
   # had to custom implement this
   # any way using io??
   # yield each CSV row across keys that only provide `read()` method

我最初的尝试是尝试让botoKey遵守IOBase。我将用缓冲读取器编写它，然后尝试使用TextIOWrapper读取其中的行，但遇到了readinto的编码问题。

class IOCompatibleKey(object):

   def __init__(self, s3_key):
      self.s3_key = s3_key

   def readable(self):
      return True

   def writeable(self):
      return False

   def read(num_bytes):
      return self.s3_key.read(num_bytes)

   def readinto(n):
      # .... ?????

buffered_reader = BufferedReader(IOCompatibleKey(s3_key))
text_reader = TextIOWrapper(buffered_reader)
for line in text_reader: # <- IS THIS POSSIBLE????
    print(line)




Answer 1

在Python 2中，您希望避免使用TextIOWrapper对象，因为csv.reader（）对象需要一个bytestring。它无法处理unicode对象TextIOWrapper提供的。

提供IOBase实现在其他方面非常简单：

class IOCompatibleKey(object):    
    def __init__(self, s3_key):
        self.s3_key = s3_key

    def readable(self):
        return True

    def writeable(self):
        return False

    @property
    def closed(self):
        return self.s3_key.closed

    def close(self):
        self.s3_key.close()

    def read(self, num_bytes):
        return self.s3_key.read(num_bytes)

    def readinto(self, n):
        chunk = self.s3_key.read(len(n))
        read = len(chunk)
        n[:read] = chunk
        return read

并且在使用Python 2时仅使用BufferedReader：

buffered_reader = BufferedReader(IOCompatibleKey(s3_key))
csv_reader = csv.reader(buffered_reader)
for row in csv_reader:
    print(row)

在Python3上，只需在BufferedReader（）的顶部添加一个TextIOWrapper（）

Python 2中的演示，使用模拟键：

>>> import random, csv
>>> from io import BufferedReader
>>> class Key(object):
...     closed = False
...     def read(self, bytes=1024):
...         if random.random() < 0.2:
...             bytes = random.randrange(bytes)
...         return ''.join([random.choice('abcdefghijklmnopqrstuvwxyz \n,') for _ in range(bytes)])
...
>>> s3_key = Key()
>>> buffered_reader = BufferedReader(IOCompatibleKey(s3_key))
>>> next(buffered_reader)   # produces a single \n terminated line
'nffdahuitmdaktibxjsdgyhlyfm gurfyo,nt\n'
>>> reader = csv.reader(buffered_reader)  # which satisfies csv.reader
>>> next(reader)
['bi iydribq', 'u']
>>> next(reader)
['qzxtbhkk se', 'v', 'b', 'nunyjemtkxaphuqmvgfrfjdloxwohqamdtvfqgddfna cjuzpaotccenxhhhgnvrbey']