Serve file from zipfile using FastAPI

181 views Asked by At

I would like to serve a file from a zip file.

Is there some method to server files from a zip file that is nice and supports handling exceptions?


Here are my experiments

There is the first naive approach but served files can be arbitrarily large so I don't want to load the whole content in memory.

import zipfile
from typing import Annotated, Any

from fastapi import FastAPI, Depends
from fastapi.responses import StreamingResponse

app = FastAPI()

zip_file_path = "data.zip"
file_path = "index.html"

@app.get("/zip0")
async def zip0():
    with zipfile.ZipFile(zip_file_path, 'r') as zip_file:
        return Response(content=zip_file.read(file_path))

FastAPI/starlette provides StreamingResponse which should do exactly what I want but it does not work in this case with zipfile claiming read from closed file.


@app.get("/zip1")
async def zip1():
    with zipfile.ZipFile(zip_file_path, 'r') as zip_file:
        with zip_file.open(file_path) as file_like:
            return StreamingResponse(file_like)

I can do a hack by moving everything to another function and setting it as a dependency. Now I can use yield so it correctly streams the content and closes the file after finishing. The problem is that it is an ugly hack and also using Response as a dependency is not supported. Just "fixing" the type annotation from Any to StreamingResponse raises an assertion saying a big no-no to using StreamingResponse for dependency injection.

def get_file_stream_from_zip():
    with zipfile.ZipFile(zip_file_path, 'r') as zip_file:
        with zip_file.open(file_path) as file_like:
            yield StreamingResponse(file_like)


@app.get("/zip2")
async def zip2(
        streaming_response: Annotated[Any, Depends(get_file_stream_from_zip)],
):
    return streaming_response

I can do something in the middle that seems legit, but it is not possible to handle exceptions. Headers are already sent when the code recognizes that e.g. the zip file does not exist. Which is not a problem with the previous methods.

def get_file_from_zip():
    with zipfile.ZipFile(zip_file_path, 'r') as zip_file:
        with zip_file.open(file_path) as file_like:
            yield file_like


@app.get("/zip3")
async def zip3(
        file_like: Annotated[BinaryIO, Depends(get_file_from_zip)],
):
    return StreamingResponse(file_like)
0

There are 0 answers