Подтвердить что ты не робот

Адаптировать итератор, чтобы вести себя как файл-подобный объект в Python

У меня есть генератор, производящий список строк. Есть ли утилита/адаптер в Python, который может сделать его похожим на файл?

Например,

>>> 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

Поскольку данные могут быть большими и должны быть потоковыми (каждый фрагмент составляет несколько килобайт, весь поток - десятки мегабайт), я не хочу с нетерпением оценивать весь генератор, прежде чем передавать его в потоковый адаптер.

4b9b3361

ОТВЕТЫ

Ответ 1

Вот решение, которое должно читать из вашего итератора в кусках.

class some_magic_adaptor:
  def __init__( self, it ):
    self.it = it
    self.next_chunk = ""
  def growChunk( self ):
    self.next_chunk = self.next_chunk + self.it.next()
  def read( self, n ):
    if self.next_chunk == None:
      return None
    try:
      while len(self.next_chunk)<n:
        self.growChunk()
      rv = self.next_chunk[:n]
      self.next_chunk = self.next_chunk[n:]
      return rv
    except StopIteration:
      rv = self.next_chunk
      self.next_chunk = None
      return rv


def str_fn():
  for c in 'a', 'b', 'c':
    yield c * 3

ff = some_magic_adaptor( str_fn() )

while True:
  data = ff.read(4)
  if not data:
    break
  print data

Ответ 2

"Правильный" способ сделать это наследуется от стандартного базового класса Python io. Однако не кажется, что Python позволяет вам предоставить исходный текстовый класс и обернуть его буферизованным считывателем любого типа.

Лучший класс для наследования - TextIOBase. Здесь такая реализация, обработка readline и read при одновременном учете производительности. (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)

Ответ 3

Проблема с StringIO заключается в том, что вы должны загружать все в буфер вверх. Это может быть проблемой, если генератор бесконечен:)

from itertools import chain, islice
class some_magic_adaptor(object):
    def __init__(self, src):
        self.src = chain.from_iterable(src)
    def read(self, n):
        return "".join(islice(self.src, None, n))

Ответ 4

Здесь приведена модифицированная версия ответа Джона и Мэтта, которая может читать список/генератор строк и выводить bytearrays

import itertools as it
from io import TextIOBase

class IterStringIO(TextIOBase):
    def __init__(self, iterable=None):
        iterable = iterable or []
        self.iter = it.chain.from_iterable(iterable)

    def not_newline(self, s):
        return s not in {'\n', '\r', '\r\n'}

    def write(self, iterable):
        to_chain = it.chain.from_iterable(iterable)
        self.iter = it.chain.from_iterable([self.iter, to_chain])

    def read(self, n=None):
        return bytearray(it.islice(self.iter, None, n))

    def readline(self, n=None):
        to_read = it.takewhile(self.not_newline, self.iter)
        return bytearray(it.islice(to_read, None, n))

использование:

ff = IterStringIO(c * 3 for c in ['a', 'b', 'c'])

while True:
    data = ff.read(4)

    if not data:
        break

    print data

aaab
bbcc
c

альтернативное использование:

ff = IterStringIO()
ff.write('ddd')
ff.write(c * 3 for c in ['a', 'b', 'c'])

while True:
    data = ff.read(4)

    if not data:
        break

    print data

ddda
aabb
bccc

Ответ 5

Существует один под названием werkzeug.contrib.iterio.IterIO, но обратите внимание, что он хранит весь итератор в своей памяти (вплоть до того момента, когда вы прочитали его как файл), поэтому он может быть непригоден.

http://werkzeug.pocoo.org/docs/contrib/iterio/

Источник: https://github.com/mitsuhiko/werkzeug/blob/master/werkzeug/contrib/iterio.py

Открытая ошибка на readline/iter: https://github.com/mitsuhiko/werkzeug/pull/500

Ответ 6

это именно то, для чего используется stringIO..

>>> import StringIO
>>> some_var = StringIO.StringIO("Hello World!")
>>> some_var.read(4)
'Hell'
>>> some_var.read(4)
'o Wo'
>>> some_var.read(4)
'rld!'
>>>

Или, если вы хотите сделать то, что звучит как

Class MyString(StringIO.StringIO):
     def __init__(self,*args):
         StringIO.StringIO.__init__(self,"".join(args))

тогда вы можете просто

xx = MyString(*list_of_strings)

Ответ 7

Глядя на Мэтта, я вижу, что не всегда необходимо реализовать все методы чтения. read1 может быть достаточно, что описано как:

Чтение и возврат к размеру байтов, причем не более одного вызова основных сырых потоков read()...

Затем его можно обернуть с помощью io.TextIOWrapper, который, например, имеет реализацию readline. В качестве примера здесь приводится потоковая передача CSV файла из S3 (Amazon Simple Storage Service) boto.s3.key.Key, которая реализует итератор для чтения.

import io
import csv

from boto import s3


class StringIteratorIO(io.TextIOBase):

    def __init__(self, iter):
        self._iterator = iter
        self._buffer = ''

    def readable(self):
        return True

    def read1(self, n=None):
        while not self._buffer:
            try:
                self._buffer = next(self._iterator)
            except StopIteration:
                break
        result = self._buffer[:n]
        self._buffer = self._buffer[len(result):]
        return result


conn = s3.connect_to_region('some_aws_region')
bucket = conn.get_bucket('some_bucket')
key = bucket.get_key('some.csv')    

fp = io.TextIOWrapper(StringIteratorIO(key))
reader = csv.DictReader(fp, delimiter = ';')
for row in reader:
    print(row)

Update

Здесь ответ на соответствующий вопрос, который выглядит немного лучше. Он наследует io.RawIOBase и переопределяет readinto. В Python 3 это достаточно, поэтому вместо обертывания IterStream в io.BufferedReader можно обернуть его в io.TextIOWrapper. В Python 2 read1 необходим, но он может быть просто выражен, хотя readinto.

Ответ 8

Прежде всего, вашему генератору придется выводить объекты байтов. Хотя нет ничего встроенного, вы можете использовать комбинацию http://docs.python.org/library/stringio.html и itertools.chain.