如何使用FastAPI从Zip文件提供文件服务并处理异常？

我希望从Zip文件中提供文件服务，请问是否存在优雅且支持异常处理的实现方法？

我的实验尝试 #

1. 基础方案：加载全文件到内存，不适用于大文件

该方案会把整个文件内容读入内存，大文件场景下性能差。

import zipfile
from typing import Annotated, Any

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

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

2. StreamingResponse直接使用：触发"read from closed file"报错

FastAPI的StreamingResponse本应适合流式传输，但这里因为with语句会提前关闭Zip文件和内部文件句柄，导致响应流读取时文件已关闭。

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

3. 依赖注入Hack方案：可正常流式传输但不规范

把StreamingResponse放到依赖的yield中，虽然能保证文件句柄在传输完成后关闭，但属于不规范用法，且StreamingResponse本身不支持作为依赖注入类型。

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

4. 折中方案：合规但无法处理异常

依赖中返回文件句柄，再用StreamingResponse包装，但异常（比如Zip文件不存在）会在响应头发送后才抛出，导致无法正确返回错误响应。

from io import BinaryIO

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)

优雅且支持异常处理的实现方案 #

核心思路是提前完成异常检查，自定义流式迭代器管理资源，确保异常发生在响应头发送前，同时正确关闭文件句柄。

import zipfile
from typing import AsyncGenerator
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()

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

async def zip_file_streamer(zip_path: str, inner_file_path: str) -> AsyncGenerator[bytes, None]:
    zip_file = None
    inner_file = None
    try:
        zip_file = zipfile.ZipFile(zip_path, 'r')
        # 提前检查目标文件是否在Zip内
        if inner_file_path not in zip_file.namelist():
            raise HTTPException(status_code=404, detail="文件在压缩包中不存在")
        
        inner_file = zip_file.open(inner_file_path)
        # 分块读取，默认1MB块可根据需求调整
        chunk_size = 1024 * 1024
        while chunk := inner_file.read(chunk_size):
            yield chunk
    except zipfile.BadZipFile:
        raise HTTPException(status_code=400, detail="无效的压缩文件")
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail="压缩文件不存在")
    finally:
        # 确保所有文件句柄被关闭
        if inner_file:
            inner_file.close()
        if zip_file:
            zip_file.close()

@app.get("/zip")
async def serve_zip_file():
    return StreamingResponse(zip_file_streamer(zip_file_path, file_path))

方案优势：

• 流式传输：分块读取内容，不会占用大量内存，适配大文件场景
• 完整异常处理：在响应头发送前完成所有前置检查，能返回标准HTTP错误响应
• 安全资源管理：通过finally块确保文件句柄被关闭，避免资源泄漏
• 规范实现：遵循FastAPI响应流设计，无依赖注入的不规范用法

内容的提问来源于stack exchange，提问作者j123b567