Last active
October 27, 2021 16:28
-
-
Save versusvoid/ae88f495a21e333ff53f781466561206 to your computer and use it in GitHub Desktop.
Flask stream generated .zip file
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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