Provide an example with Fast API

[SeeRich]

Is there any way you could provide an example using Fast APIs StreamingResponse with some files?

The following code is a standalone example using Python's zipfile package. However, I would love to be able to use this package instead of blocking the endpoint while the zipfile is created (here as BytesIO object).

You will need to pip install [uvicorn, aiofiles, fastapi]

Thank you! Let me know if there are any questions from the example.

import zipfile
from pathlib import Path
from io import BytesIO
from typing import List

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

# Run with: uvicorn test:app --reload

app = FastAPI()


@app.get("/download")
async def main():
    # Collect the file names you are interested in zipping up
    cwd = Path().cwd()
    filepaths: List[Path] = [
        cwd / "testdata" / "test2.json",
        cwd / "testdata" / "test.json"
    ]

    zip_io = BytesIO()
    with zipfile.ZipFile(zip_io, mode="w",
                         compression=zipfile.ZIP_DEFLATED, compresslevel=0) as temp_zip:
        for fp in filepaths:
            if not fp.exists():
                continue
            temp_zip.write(str(fp), fp.name)
    return StreamingResponse(
        iter([zip_io.getvalue()]),
        media_type="application/x-zip-compressed",
        headers={"Content-Disposition": "attachment; filename=archive.zip"})     

[michalc]

So I've not tested it, but I think the return value of stream_zip should be able to be passed as the first argument to StreamingResponse:

from datetime import datetime
from stream_zip import ZIP_32, stream_zip

@app.get("/download")
async def main():
    cwd = Path().cwd()
    filepaths: List[Path] = [
        cwd / "testdata" / "test2.json",
        cwd / "testdata" / "test.json"
    ]
    now = datetime.now()

    def file_contents(fp):
        while chunk := fp.read(65536):
            yield chunk

    def member_files():
        for fp in filepaths:
            if not fp.exists():
                continue
            yield (fp.name, now, 0o600, ZIP_32, file_contents(fp))

    return StreamingResponse(
        stream_zip(member_files()),
        media_type="application/x-zip-compressed",
        headers={"Content-Disposition": "attachment; filename=archive.zip"})  

Note: when in a FastAPI async endpoint like this, looks like under the hood it runs StreamingResponse iteration in a thread via https://anyio.readthedocs.io/en/stable/api.html#running-code-in-worker-threads, so some of the atomicity guarantees that you might be used to if everything is pure asyncio might not hold. I suspect it's fine in the above example since nothing is getting mutated in the member_files function and there isn't another thread running stuff, but if it were, you might have to be careful

[SeeRich]

Thank you for example. This works great! I will post a full example using FastAPI + requests (for testing) if you are interested in including it in your documentation.

[SeeRich]

Full example of the solution for downloading Zip file within FastAPI framework:

import argparse
from datetime import datetime
from pathlib import Path
from typing import List
import zlib

# pip install uvicorn requests fastapi stream-zip
import uvicorn
import requests
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from stream_zip import ZIP_64, stream_zip

# Run the api using `python filename.py`
# Run the download test using `python filename.py -d`

app = FastAPI()

MEBIBYTE = 1024 * 1024


def sizeof_fmt(bytes, suffix="B"):
    for unit in ["", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi"]:
        if abs(bytes) < 1024.0:
            return f"{bytes:3.1f}{unit}{suffix}"
        bytes /= 1024.0
    return f"{bytes:.1f}Yi{suffix}"


def file_timestamp(fp: Path) -> datetime:
    return datetime.fromtimestamp(fp.stat().st_ctime)


def download_file(url, local_filepath: Path):
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        total_bytes_received = 0
        prev_bytes_printed = 0
        with open(str(local_filepath), 'wb') as f:
            for chunk in r.iter_content(chunk_size=8192):
                total_bytes_received += f.write(chunk)
                # Only print the message every Ten MEBIBYTES
                if total_bytes_received - prev_bytes_printed > 10 * MEBIBYTE:
                    print(f"Cumulative Bytes Received: {sizeof_fmt(total_bytes_received)}")
                    prev_bytes_printed = total_bytes_received
    print(f"Downloaded ({sizeof_fmt(total_bytes_received)}): {local_filepath}")


@app.get("/download")
async def main():
    # Collect the file names you are interested in zipping up
    cwd = Path().cwd()
    filepaths: List[Path] = [
        cwd / "testdata" / "test.mp4",
        cwd / "testdata" / "test.json",
        cwd / "testdata" / "test2.json"
    ]

    def file_contents(fp):
        with open(str(fp), "rb") as fo:
            while chunk := fo.read(65536):
                yield chunk

    def member_files():
        for fp in filepaths:
            if not fp.exists():
                continue
            yield (fp.name, file_timestamp(fp), 0o600, ZIP_64, file_contents(fp))

    return StreamingResponse(
        stream_zip(member_files(),
                   get_compressobj=lambda: zlib.compressobj(level=0)),
        media_type="application/x-zip-compressed",
        headers={"Content-Disposition": "attachment; filename=archive.zip"})


if __name__ == "__main__":
    p = argparse.ArgumentParser(
        description="Example of streaming download endpoint using the Python FastAPI framework")
    p.add_argument("-d", "--download", action="store_true",
                   dest="download", default=False,
                   help="Run the download function instead of FastAPI")
    args = p.parse_args()

    if not args.download:
        # Launch FastAPI using uvicorn
        uvicorn.run(app, host="127.0.0.1", port=5000, log_level="info")
    else:
        # Call the download function
        download_file("http://127.0.0.1:5000/download",
                      Path().cwd() / "archive.zip")

[michalc]

Having something in the docs sounds good,

Although I'm a bit tempted that anything in the docs be quite a brief output recipe. So something like:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from stream_zip import stream_zip

app = FastAPI()

# Define a generator function "member_files" that yields the member files of the zip
def member_files():
    ...

@app.get("/the/endpoint")
async def main():
    return StreamingResponse(member_files(), {
        media_type="application/zip",
        headers={"content-disposition": "attachment; filename=archive.zip"})
    })

Essentially since where the member_files comes from is a very case-by-case basis