使用 python io 从缓冲流中组成一个行阅读器

Question

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

我想知道是否有人可以编写 python 的 io 类来实现这一点？目标是拥有某种能够包装 boto 密钥的抽象，并在密钥上提供readline 或迭代器接口（仅提供read(size=0) 调用。复杂性在于，因为它存储为 CSV ，每一行都是可变长度的。

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

看起来 python io 真的有所有的东西可以做到这一点BufferedReader 和 TextIOWrapper，我天真地试图将 boto Key 传递给它，但 BufferedReader 预期一个IOBase 对象。

然后我围绕 Key 实现了 IOBase 协议，但出现了 unicode 错误，并且通常不确定我在做什么。

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

---

技术规格：

s3上有1-100个CSV文件的目录。都具有相同的格式，但行数可变。我正在尝试实现一个函数，该函数采用 boto Keys 的迭代器。

Key 提供了一个read(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

---

我最初的尝试是尝试让 boto Key 遵守 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() 对象需要一个字节串。它无法处理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)

在 Python 3 上，只需在 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']