Stream a non-seekable file-like object to multiple sinks
<p>I have a non-seekable file-like object. In particular it is a file of indeterminate size coming from an HTTP request.</p>

<pre><code>import requests
fileobj = requests.get(url, stream=True)
</code></pre>

<p>I am streaming this file to a call to an Amazon AWS SDK function which is writing the contents to Amazon S3. This is working fine.</p>

<pre><code>import boto3
s3 = boto3.resource('s3')
s3.bucket('my-bucket').upload_fileobj(fileobj, 'target-file-name')
</code></pre>

<p>However, at the same time as streaming it to S3 I want to also stream the data to another process. This other process may not need the entire stream and might stop listening at some point; this is fine and should not affect the stream to S3.</p>

<p>It's important I don't use too much memory, since some of these files could be enormous. I don't want to write anything to disk for the same reason.</p>

<p>I don't mind if either sink is slowed down due to the other being slow, as long as S3 eventually gets the entire file, and the data goes to both sinks (rather, to each one which still wants it).</p>

<p>What's the best way to go about this in Python (3)? I know I can't just pass the same file object to both sinks, such as</p>

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

<p>I think I could write a wrapper for the file-like object which passes any data read not only to the caller (which would be the S3 sink) but also to the other process. Something like</p>

<pre><code>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')
</code></pre>

<p>But is there a better way to do it? Perhaps something like</p>

<pre><code>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])
</code></pre>

<p>The discomfort regarding your <code>MyFilewrapper</code> solution arises, because the IO loop inside <code>upload_fileobj</code> is now in control of feeding the data to a subprocess that is strictly speaking unrelated to the upload.</p>

<p>A "proper" solution would involve an upload API that provides a file-like object for <strong>writing</strong> the upload stream with an outside loop. That would allow you to feed the data to both target streams "cleanly".</p>

<p>The following example shows the basic concept. The fictional <code>startupload</code> method provides the file-like object for uploading. Of cource you would need to add proper error handling etc.</p>

<pre><code>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)
</code></pre>

python python-3.x stream