Adapt an iterator to behave like a file-like object in Python

Question

I have a generator producing a list of strings. Is there a utility/adapter in Python that could make it look like a file?

For example,

>>> def str_fn(): ...     for c in 'a', 'b', 'c': ...         yield c * 3 ...  >>> for s in str_fn(): ...     print s ...  aaa bbb ccc >>> stream = some_magic_adaptor(str_fn()) >>> while True: ...    data = stream.read(4) ...    if not data: ...        break ...    print data aaab bbcc c 

Because data may be big and needs to be streamable (each fragment is a few kilobytes, the entire stream is tens of megabytes), I do not want to eagerly evaluate the whole generator before passing it to stream adaptor.

Answer 1

The "correct" way to do this is inherit from a standard Python io abstract base class. However it doesn't appear that Python allows you to provide a raw text class, and wrap this with a buffered reader of any kind.

The best class to inherit from is TextIOBase. Here's such an implementation, handling readline, and read while being mindful of performance. (gist)

import io  class StringIteratorIO(io.TextIOBase):      def __init__(self, iter):         self._iter = iter         self._left = ''      def readable(self):         return True      def _read1(self, n=None):         while not self._left:             try:                 self._left = next(self._iter)             except StopIteration:                 break         ret = self._left[:n]         self._left = self._left[len(ret):]         return ret      def read(self, n=None):         l = []         if n is None or n < 0:             while True:                 m = self._read1()                 if not m:                     break                 l.append(m)         else:             while n > 0:                 m = self._read1(n)                 if not m:                     break                 n -= len(m)                 l.append(m)         return ''.join(l)      def readline(self):         l = []         while True:             i = self._left.find('\n')             if i == -1:                 l.append(self._left)                 try:                     self._left = next(self._iter)                 except StopIteration:                     self._left = ''                     break             else:                 l.append(self._left[:i+1])                 self._left = self._left[i+1:]                 break         return ''.join(l)