zipstream-new

import zipstream import os def generate_zip_stream(): # Create a dummy directory and files for demonstration if not os.path.exists('test_files'): os.makedirs('test_files') with open('test_files/file1.txt', 'w') as f: f.write('This is file 1 content.') with open('test_files/file2.txt', 'w') as f: f.write('This is file 2 content. More data...') # Initialize ZipFile for streaming # Use allowZip64=True for archives larger than 4GB or 65,535 files # Use compression=zipstream.ZIP_DEFLATED for compressed output zs = zipstream.ZipFile(mode='w', compression=zipstream.ZIP_DEFLATED, allowZip64=True) # Add files from a path zs.write('test_files/file1.txt', arcname='archive/document1.txt') # Add content from a string or bytes directly zs.writestr('archive/dynamic_content.json', b'{"data": [1, 2, 3]}') # Add content from an iterable (e.g., a generator) def data_generator(): yield b"Line 1\n" yield b"Line 2\n" yield b"Line 3\n" zs.write_iter('archive/generated_log.txt', data_generator()) # Iterate over the zipstream to get chunks of the ZIP file # These chunks can be streamed directly to a HTTP response or cloud storage for chunk in zs: yield chunk # Clean up dummy files/directory (optional) os.remove('test_files/file1.txt') os.remove('test_files/file2.txt') os.rmdir('test_files') # Example of consuming the stream to a local file with open('output.zip', 'wb') as f: for chunk in generate_zip_stream(): f.write(chunk) print("ZIP file 'output.zip' created successfully.")