如何让REST接口同时返回JSON payload与数据集压缩包？

HTTP响应默认只能有一种Content-Type，没法直接在同一个响应里同时返回JSON和二进制压缩包，不过有几种实用方案可以满足你的需求，根据场景选择即可：

方案1：将JSON元数据嵌入压缩包 #

把生成的JSON元数据作为单独文件（比如metadata.json）放进压缩包，返回整个zip文件。客户端解压后就能同时拿到数据文件和完整的元数据。

代码示例：

from fastapi import Response
from zipfile import ZipFile
import io
import requests
from your_module import DummyDataResponseBodySuccess

# 假设你已经生成了符合模型的数据集对象dummy_datasets
# 生成JSON元数据字符串
metadata_json = dummy_datasets.model_dump_json(indent=2)

# 创建内存中的zip流
zip_stream = io.BytesIO()
with ZipFile(zip_stream, 'w') as zip_file:
    # 添加元数据文件
    zip_file.writestr("metadata.json", metadata_json)
    # 遍历数据集，添加所有数据文件（这里示例从URL拉取内容，实际根据你的逻辑调整）
    for item in dummy_datasets.dataset:
        file_content = requests.get(item.location).content
        zip_file.writestr(item.name, file_content)

# 将流指针移到开头
zip_stream.seek(0)

# 返回zip响应，添加下载文件名头
return Response(
    content=zip_stream.getvalue(),
    media_type="application/zip",
    headers={"Content-Disposition": "attachment; filename=dataset_with_metadata.zip"}
)

方案2：返回JSON+压缩包下载链接 #

保留原有JSON响应结构，新增一个zip_download_url字段指向生成好的压缩包地址。这种方式对客户端最友好，不需要特殊解析逻辑。

首先修改Pydantic模型：

from pydantic import BaseModel, Field, HttpUrl

class ResponseBodySuccessDataset(BaseModel):
    name: str = Field(description="Name of the generated data object")
    location: HttpUrl = Field(
        description="URL of generated data object as returned to requestor",
    )

class DummyDataResponseBodySuccess(BaseModel):
    dataset: list[ResponseBodySuccessDataset] = Field(
        description="Dataset array as returned to requestor",
    )
    zip_download_url: HttpUrl = Field(description="URL to download the zip package of all dataset files")

然后实现接口逻辑：

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from your_module import DummyDataResponseBodySuccess
import os
import uuid
from zipfile import ZipFile
import requests

app = FastAPI()

# 本地存储zip文件的目录，也可以替换为云对象存储（如S3）
ZIP_STORAGE_DIR = "./zip_storage"
os.makedirs(ZIP_STORAGE_DIR, exist_ok=True)

@app.get("/your-endpoint", response_model=DummyDataResponseBodySuccess)
def get_dataset():
    # 生成原有数据集（你的业务逻辑）
    dummy_datasets = DummyDataResponseBodySuccess(
        dataset=[
            {"name": "file1", "location": "https://container/file1.csv"},
            {"name": "file2", "location": "https://container/file2.csv"}
        ],
        zip_download_url=""  # 先占位，后面赋值
    )

    # 生成唯一zip文件名，避免冲突
    zip_filename = f"dataset_{uuid.uuid4()}.zip"
    zip_file_path = os.path.join(ZIP_STORAGE_DIR, zip_filename)
    
    # 生成zip文件并保存
    with ZipFile(zip_file_path, 'w') as zip_file:
        for item in dummy_datasets.dataset:
            file_content = requests.get(item.location).content
            zip_file.writestr(item.name, file_content)

    # 生成压缩包的下载URL（根据你的服务域名调整）
    zip_download_url = f"http://your-service-domain/download-zip/{zip_filename}"

    # 返回带下载链接的完整JSON
    return DummyDataResponseBodySuccess(
        dataset=dummy_datasets.dataset,
        zip_download_url=zip_download_url
    )

# 新增压缩包下载接口
@app.get("/download-zip/{zip_filename}")
def download_zip(zip_filename: str):
    zip_file_path = os.path.join(ZIP_STORAGE_DIR, zip_filename)
    if not os.path.exists(zip_file_path):
        raise HTTPException(status_code=404, detail="Zip file not found")
    return FileResponse(zip_file_path, media_type="application/zip", filename=zip_filename)

方案3：使用Multipart响应（不推荐，客户端处理复杂） #

通过HTTP的Multipart格式，在一个响应里拆分出JSON和zip两个部分。但这种方式需要客户端能解析多部分响应，兼容性不如前两种，仅适合特定场景。

代码示例：

from fastapi import Response
import io
from zipfile import ZipFile
import requests
from your_module import DummyDataResponseBodySuccess

def build_multipart_response(json_bytes: bytes, zip_bytes: bytes):
    # 定义分隔符，确保不会和内容冲突
    boundary = b"custom-multipart-boundary-789"
    # 构造多部分内容
    parts = [
        b"--" + boundary,
        b'Content-Type: application/json',
        b'',
        json_bytes,
        b"--" + boundary,
        b'Content-Type: application/zip',
        b'Content-Disposition: attachment; filename=dataset.zip',
        b'',
        zip_bytes,
        b"--" + boundary + b"--",
    ]
    return b"\r\n".join(parts)

@app.get("/your-endpoint")
def get_dataset():
    # 生成数据集
    dummy_datasets = DummyDataResponseBodySuccess(
        dataset=[
            {"name": "file1", "location": "https://container/file1.csv"},
            {"name": "file2", "location": "https://container/file2.csv"}
        ]
    )
    json_bytes = dummy_datasets.model_dump_json().encode("utf-8")

    # 生成zip数据
    zip_stream = io.BytesIO()
    with ZipFile(zip_stream, 'w') as zip_file:
        for item in dummy_datasets.dataset:
            file_content = requests.get(item.location).content
            zip_file.writestr(item.name, file_content)
    zip_stream.seek(0)
    zip_bytes = zip_stream.getvalue()

    # 构造并返回Multipart响应
    multipart_content = build_multipart_response(json_bytes, zip_bytes)
    return Response(
        content=multipart_content,
        media_type=f"multipart/mixed; boundary=custom-multipart-boundary-789"
    )

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