AsyncZipstream

https://pypi.org/project/asynczipstream/
Admin
AsyncZipstream is an asynchronous ZIP file generator for Python that streams files and iterables...
Tokens:956
Snippets:11
Trust Score:10
Update:6 months ago
Show doc for...
Tokens:
Raw
### Install asynczipstream

Source: https://pypi.org/project/asynczipstream

Command to install the package via pip.

```bash
pip install asynczipstream
```

--------------------------------

### Integrate with aiohttp

Source: https://pypi.org/project/asynczipstream

Example of streaming a zip archive directly through an aiohttp web response.

```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)
```

--------------------------------

### Write files to a zip archive

Source: https://pypi.org/project/asynczipstream

Demonstrates writing files from a local path to a zip archive using an asynchronous iterator.

```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())
```

--------------------------------

### Combine file and iterable inputs

Source: https://pypi.org/project/asynczipstream

Combines both file-based and iterable-based inputs into a single zip archive.

```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())
```

--------------------------------

### Write byte iterables to a zip archive

Source: https://pypi.org/project/asynczipstream

Shows how to add data from a byte string iterable to a zip archive without storing the full content in memory.

```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())
```
Tokens:
Raw