Skip to content

Instantly share code, notes, and snippets.

@versusvoid
Last active October 27, 2021 16:28
Show Gist options
  • Save versusvoid/ae88f495a21e333ff53f781466561206 to your computer and use it in GitHub Desktop.
Save versusvoid/ae88f495a21e333ff53f781466561206 to your computer and use it in GitHub Desktop.
Flask stream generated .zip file
import zipfile
class StreamedBytesIO(object):
def __init__(self):
self._b = bytearray()
self._pos = 0
def seek(self, *args):
raise AttributeError('sorry')
def tell(self):
raise AttributeError('sorry')
def flush(self, *args):
pass
def close(self, *args):
pass
def write(self, data):
if self._pos + len(data) > len(self._b):
self._b.extend(0 for _ in range(max(len(data), len(self._b))))
self._b[self._pos:self._pos + len(data)] = data
self._pos += len(data)
return len(data)
def get_chunk_and_reset(self):
result = memoryview(self._b)[:self._pos].tobytes()
self._pos = 0
return result
def generate():
yield b'' # flushing response status and headers to client
b = StreamedBytesIO()
archive = zipfile.ZipFile(b, mode='w', compression=zipfile.ZIP_DEFLATED)
for path_in_archive, data in get_a_lot_of_files():
archive.writestr(path_in_archive, data)
yield b.get_chunk_and_reset()
archive.close()
yield b.get_chunk_and_reset()
@app.route('/large.zip')
def generate_large_zip():
return Response(generate(), mimetype='application/zip')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment