在 Python 中支持从 http 响应中查找的非阻塞流

Question

我有一个来自 urllib 的 http 响应

response = urllib2.urlopen('http://python.org/')

最终，我希望能够在响应中seek()（至少在开始时）。所以我希望能够有这样的代码：

print result.readline()
result.seek(0)
print result.readline()

这个问题最简单的解决方案是StringIO 或io.BytesIO 像这样：

result = io.BytesIO(response.read())

但是，问题是我想要请求的资源往往非常大，我想在整个下载完成之前开始使用它们（解析...）。 response.read() 正在阻塞。我正在寻找一个非阻塞的解决方案。

理想的代码是来自资源的read(BUFFER_SIZE)，当需要更多内容时，只需从响应中请求更多内容。我基本上是在寻找一个可以做到这一点的包装类。哦，我需要一个像对象这样的文件。

我想，我可以这样写：

base = io.BufferedIOBase(response)
result = io.BufferedReader(base)

但是，事实证明这不起作用，我尝试了与 io module 不同的类，但无法使其正常工作。我对任何具有所需行为的包装类感到满意。

Answer 1

我编写了自己的包装类来保存第一块数据。这样我可以回到开头，分析编码，文件类型和其他东西。这个类为我解决了这个问题，并且应该足够简单以适应其他用例。

class BufferedFile(object):
    ''' A buffered file that preserves the beginning of a stream up to buffer_size
    '''
    def __init__(self, fp, buffer_size=1024):
        self.data = cStringIO.StringIO()
        self.fp = fp
        self.offset = 0
        self.len = 0
        self.fp_offset = 0
        self.buffer_size = buffer_size

    @property
    def _buffer_full(self):
        return self.len >= self.buffer_size

    def readline(self):
        if self.len < self.offset < self.fp_offset:
            raise BufferError('Line is not available anymore')
        if self.offset >= self.len:
            line = self.fp.readline()
            self.fp_offset += len(line)

            self.offset += len(line)

            if not self._buffer_full:
                self.data.write(line)
                self.len += len(line)
        else:
            line = self.data.readline()
            self.offset += len(line)
        return line

    def seek(self, offset):
        if self.len < offset < self.fp_offset:
            raise BufferError('Cannot seek because data is not buffered here')
        self.offset = offset
        if offset < self.len:
            self.data.seek(offset)

Answer 2

使用Requests 库，您可以迭代响应，即streamed：

要使用 Twitter 流 API 来跟踪关键字“请求”：

import requests
import json

r = requests.post('https://stream.twitter.com/1/statuses/filter.json',
    data={'track': 'requests'}, auth=('username', 'password'), stream=True)

for line in r.iter_lines():
    if line: # filter out keep-alive new lines
        print json.loads(line)

为了能够搜索，您必须保存已经迭代（读取）的数据。