# python-zipstream-aio
zipstream.py is a zip archive generator based on python 3.3's zipfile.py. It was created to
generate a zip file generator for streaming (ie web apps). This is beneficial for when you
want to provide a downloadable archive of a large collection of regular files, which would be infeasible to
generate the archive prior to downloading or of a very large file that you do not want to store entirely on disk or on memory.
The archive is generated as an iterator of strings, which, when joined, form
the zip archive. For example, the following code snippet would write a zip
archive containing files from 'path' to a normal file:
```python
import asynczipstream
import asyncio
async def main():
z = asynczipstream.ZipFile()
z.write('path/to/files')
with open('zipfile.zip', 'wb') as f:
async for data in z:
f.write(data)
asyncio.run(main())
```
zipstream also allows to take as input a byte string iterable and to generate
the archive as an iterator.
This avoids storing large files on disk or in memory.
To do so you could use something like this snippet:
```python
import asynczipstream
import asyncio
async def main():
async def iterable():
for _ in range(10):
yield b'this is a byte string\x01\n'
z = asynczipstream.ZipFile()
z.write_iter('my_archive_iter', iterable())
with open('zipfile.zip', 'wb') as f:
async for data in z:
f.write(data)
asyncio.run(main())
```
Of course both approach can be combined:
```python
import asynczipstream
import asyncio
async def main():
async def iterable():
for _ in range(10):
yield b'this is a byte string\x01\n'
z = asynczipstream.ZipFile()
z.write('path/to/files', 'my_archive_files')
z.write_iter('my_archive_iter', iterable())
with open('zipfile.zip', 'wb') as f:
async for data in z:
f.write(data)
asyncio.run(main())
```
If the zlib module is available, asynczipstream.ZipFile can generate compressed zip
archives.
## Installation
```
pip install asynczipstream
```
## Requirements
* Python 3.8+
### aiohttp Example
```python
from aiohttp import web, hdrs
import asynczipstream
async def handle_license(request):
"""
Example with file from disk
"""
filename = "license.zip"
response = web.StreamResponse(
status=200,
headers={
hdrs.CONTENT_TYPE: "application/octet-stream",
hdrs.CONTENT_DISPOSITION: (f"attachment; " f'filename="{filename}"; '),
},
)
await response.prepare(request)
z = asynczipstream.ZipFile()
z.write('LICENSE')
async for data in z:
await response.write(data)
return response
async def handle_readme(request):
"""
Example with file from iterable object
"""
filename = "readme.zip"
response = web.StreamResponse(
status=200,
headers={
hdrs.CONTENT_TYPE: "application/octet-stream",
hdrs.CONTENT_DISPOSITION: (f"attachment; " f'filename="{filename}"; '),
},
)
await response.prepare(request)
async def iterable(filename):
with open(filename, "rb") as f:
yield f.readline()
z = asynczipstream.ZipFile()
z.write_iter('README.md', iterable("README.md"))
async for data in z:
await response.write(data)
return response
app = web.Application()
app.add_routes([web.get('/license', handle_license),web.get('/readme', handle_readme)])
if __name__ == '__main__':
web.run_app(app)
```
## Running tests
Just run the following command: `python -m unittest discover`
Raw data
{
"_id": null,
"home_page": "https://github.com/lososkin/python-zipstream-aio",
"name": "asynczipstream",
"maintainer": "",
"docs_url": null,
"requires_python": "",
"maintainer_email": "",
"keywords": "async zip streaming",
"author": "lososkin",
"author_email": "yalososka@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/df/c4/3d094bc654226c0a7eac6c519d75e819e8743458a8e7dbb4b1d00d932f4e/asynczipstream-1.0.1.tar.gz",
"platform": null,
"description": "\n# python-zipstream-aio\n\nzipstream.py is a zip archive generator based on python 3.3's zipfile.py. It was created to\ngenerate a zip file generator for streaming (ie web apps). This is beneficial for when you\nwant to provide a downloadable archive of a large collection of regular files, which would be infeasible to\ngenerate the archive prior to downloading or of a very large file that you do not want to store entirely on disk or on memory.\n\nThe archive is generated as an iterator of strings, which, when joined, form\nthe zip archive. For example, the following code snippet would write a zip\narchive containing files from 'path' to a normal file:\n\n```python\nimport asynczipstream\nimport asyncio\n\nasync def main():\n z = asynczipstream.ZipFile()\n z.write('path/to/files')\n\n with open('zipfile.zip', 'wb') as f:\n async for data in z:\n f.write(data)\n\nasyncio.run(main())\n```\n\nzipstream also allows to take as input a byte string iterable and to generate\nthe archive as an iterator.\nThis avoids storing large files on disk or in memory.\nTo do so you could use something like this snippet:\n\n```python\nimport asynczipstream\nimport asyncio\n\nasync def main():\n async def iterable():\n for _ in range(10):\n yield b'this is a byte string\\x01\\n'\n\n z = asynczipstream.ZipFile()\n z.write_iter('my_archive_iter', iterable())\n\n with open('zipfile.zip', 'wb') as f:\n async for data in z:\n f.write(data)\n\nasyncio.run(main())\n```\n\nOf course both approach can be combined:\n\n```python\nimport asynczipstream\nimport asyncio\n\nasync def main():\n async def iterable():\n for _ in range(10):\n yield b'this is a byte string\\x01\\n'\n\n z = asynczipstream.ZipFile()\n z.write('path/to/files', 'my_archive_files')\n z.write_iter('my_archive_iter', iterable())\n\n with open('zipfile.zip', 'wb') as f:\n async for data in z:\n f.write(data)\n\nasyncio.run(main())\n```\n\nIf the zlib module is available, asynczipstream.ZipFile can generate compressed zip\narchives.\n\n## Installation\n\n```\npip install asynczipstream\n```\n\n## Requirements\n\n * Python 3.8+\n\n\n### aiohttp Example\n\n```python\nfrom aiohttp import web, hdrs\nimport asynczipstream\n\nasync def handle_license(request):\n \"\"\"\n Example with file from disk\n \"\"\"\n filename = \"license.zip\"\n response = web.StreamResponse(\n status=200,\n headers={\n hdrs.CONTENT_TYPE: \"application/octet-stream\",\n hdrs.CONTENT_DISPOSITION: (f\"attachment; \" f'filename=\"{filename}\"; '),\n },\n )\n await response.prepare(request)\n\n z = asynczipstream.ZipFile()\n z.write('LICENSE')\n async for data in z:\n await response.write(data)\n return response\n\nasync def handle_readme(request):\n \"\"\"\n Example with file from iterable object\n \"\"\"\n filename = \"readme.zip\"\n response = web.StreamResponse(\n status=200,\n headers={\n hdrs.CONTENT_TYPE: \"application/octet-stream\",\n hdrs.CONTENT_DISPOSITION: (f\"attachment; \" f'filename=\"{filename}\"; '),\n },\n )\n await response.prepare(request)\n\n async def iterable(filename):\n with open(filename, \"rb\") as f:\n yield f.readline()\n\n z = asynczipstream.ZipFile()\n z.write_iter('README.md', iterable(\"README.md\"))\n async for data in z:\n await response.write(data)\n return response\n\n\napp = web.Application()\napp.add_routes([web.get('/license', handle_license),web.get('/readme', handle_readme)])\n\nif __name__ == '__main__':\n web.run_app(app)\n```\n\n## Running tests\n\nJust run the following command: `python -m unittest discover`\n",
"bugtrack_url": null,
"license": "",
"summary": "Asynchronous Zipfile generator that takes input files as well as streams",
"version": "1.0.1",
"project_urls": {
"Homepage": "https://github.com/lososkin/python-zipstream-aio"
},
"split_keywords": [
"async",
"zip",
"streaming"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "12ae83f33d0c5e02d223115ceeea4c9f4041b57000ba2f0815da692a7464d75c",
"md5": "9b3c5e485774fdaa28afe432b30c9664",
"sha256": "a57f78fb36054450f57472f3f10b3b15ba93cd2547a0055a1e7b08a78f82674d"
},
"downloads": -1,
"filename": "asynczipstream-1.0.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "9b3c5e485774fdaa28afe432b30c9664",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 20187,
"upload_time": "2023-05-13T20:48:21",
"upload_time_iso_8601": "2023-05-13T20:48:21.348742Z",
"url": "https://files.pythonhosted.org/packages/12/ae/83f33d0c5e02d223115ceeea4c9f4041b57000ba2f0815da692a7464d75c/asynczipstream-1.0.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "dfc43d094bc654226c0a7eac6c519d75e819e8743458a8e7dbb4b1d00d932f4e",
"md5": "775268349512c40af8e755fa254a0d1d",
"sha256": "800fa0d9849e8ea46dcac8f283b0efe70d86e0f5a7c441c1896a068b94c3c2ed"
},
"downloads": -1,
"filename": "asynczipstream-1.0.1.tar.gz",
"has_sig": false,
"md5_digest": "775268349512c40af8e755fa254a0d1d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 22301,
"upload_time": "2023-05-13T20:48:23",
"upload_time_iso_8601": "2023-05-13T20:48:23.172808Z",
"url": "https://files.pythonhosted.org/packages/df/c4/3d094bc654226c0a7eac6c519d75e819e8743458a8e7dbb4b1d00d932f4e/asynczipstream-1.0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-05-13 20:48:23",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "lososkin",
"github_project": "python-zipstream-aio",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "asynczipstream"
}
JSON
