python - 将不可搜索的类文件对象流式传输到多个接收器

Question

我有一个不可搜索的类文件对象。特别是它是来自 HTTP 请求的不确定大小的文件。

import requests
fileobj = requests.get(url, stream=True)

我正在将此文件流式传输到对 Amazon AWS SDK 函数的调用，该函数将内容写入 Amazon S3。这工作正常。

import boto3
s3 = boto3.resource('s3')
s3.bucket('my-bucket').upload_fileobj(fileobj, 'target-file-name')

但是，在将其流式传输到 S3 的同时，我还想将数据流式传输到另一个进程。另一个进程可能不需要整个流，并且可能会在某个时候停止监听；这很好，不应该影响到 S3 的流。

重要的是我不要使用太多内存，因为其中一些文件可能很大。出于同样的原因，我不想将任何内容写入磁盘。

我不介意任何一个接收器是否因另一个接收器变慢而变慢，只要 S3 最终获取整个文件，并且数据进入两个接收器(而是发送到仍然需要它的每个接收器)。

在 Python (3) 中解决这个问题的最佳方法是什么？我知道我不能将相同的文件对象传递给两个接收器，例如

s3.bucket('my-bucket').upload_fileobj(fileobj, 'target-file-name')
# At the same time somehow as
process = subprocess.Popen(['myapp'], stdin=fileobj)

我想我可以为类似文件的对象编写一个包装器，它将读取的任何数据不仅传递给调用者(这将是 S3 接收器)，而且传递给另一个进程。就像是

class MyFilewrapper(object):
    def __init__(self, fileobj):
        self._fileobj = fileobj
        self._process = subprocess.Popen(['myapp'], stdin=popen.PIPE)
    def read(self, size=-1):
        data = self._fileobj.read(size)
        self._process.stdin.write(data)
        return data

filewrapper = MyFilewrapper(fileobj)
s3.bucket('my-bucket').upload_fileobj(filewrapper, 'target-file-name')

但是有没有更好的方法来做到这一点？也许像

streams = StreamDuplicator(fileobj, streams=2)
s3.bucket('my-bucket').upload_fileobj(streams[0], 'target-file-name')
# At the same time somehow as
process = subprocess.Popen(['myapp'], stdin=streams[1])

Answer 1

关于您的不适MyFilewrapper解决方案出现了，因为内部的IO循环upload_fileobj现在控制将数据提供给严格来说与上传无关的子进程。

“正确”的解决方案将涉及上传 API，该 API 为 提供类似文件的对象。写作 带有外部循环的上传流。这将允许您“干净地”将数据提供给两个目标流。

以下示例显示了基本概念。虚构的startupload方法提供用于上传的类文件对象。当然，您需要添加适当的错误处理等。

fileobj = requests.get(url, stream=True)

upload_fd = s3.bucket('my-bucket').startupload('target-file-name')
other_fd = ... # Popen or whatever

buf = memoryview(bytearray(4046))
while True:
    r = fileobj.read_into(buf)
    if r == 0:
        break

    read_slice = buf[:r]
    upload_fd.write(read_slice)
    other_fd.write(read_slice)