如何用FastAPI接口同时返回字典与文件?
FastAPI同时返回字典数据和文件方案
需求说明
需要通过FastAPI接口同时返回结构化字典数据(如自定义的OutData)和文件,满足:
- 外部Python应用用
requests.post调用时可分别获取数据和文件 - 在FastAPI内置文档(
/docs)中可查看结构化数据并直接下载文件
方案一:响应头传递数据 + 文件响应体(推荐)
该方案实现简单,适配Swagger文档交互,适合数据量不大的场景。
实现代码
import os import json import shutil from pathlib import Path from fastapi import FastAPI from fastapi.responses import FileResponse from pydantic import BaseModel app = FastAPI() CUR_DIR = Path(os.path.dirname(__file__)) class OutData(BaseModel): name: str value: float @app.post("/") async def run(input_data: str): # 生成结构化输出数据 out_data = OutData(name="A Name", value=10) # 打包目标文件夹为zip文件 shutil.make_archive(str(CUR_DIR), "zip", str(CUR_DIR)) zip_file = CUR_DIR.parent.joinpath(f"{CUR_DIR.name}.zip") # 创建文件响应,同时将结构化数据转为JSON字符串放入自定义响应头 file_response = FileResponse(path=zip_file, filename=zip_file.name) file_response.headers["X-Out-Data"] = json.dumps(out_data.model_dump()) return file_response
外部requests调用示例
import requests import json url = "http://127.0.0.1:8000/" response = requests.post(url, json={"input_data": "some"}) # 从响应头解析结构化数据 out_data = json.loads(response.headers["X-Out-Data"]) # 保存下载的zip文件 with open("downloaded.zip", "wb") as f: f.write(response.content) # 后续操作 print(out_data) # 1. 处理本地zip文件 # 2. 分析out_data数据
Swagger文档使用
访问http://127.0.0.1:8000/docs/,点击接口的Try it out发送请求:
- 响应头中可找到
X-Out-Data字段,查看结构化数据内容 - 浏览器会自动触发zip文件的下载
方案二:Multipart多部分响应(适合大数据/多文件场景)
如果结构化数据量较大,或需要返回多个文件,可采用Multipart格式拆分内容。
实现代码
import os import json import shutil from pathlib import Path from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel from multipart.multipart import MultipartEncoder app = FastAPI() CUR_DIR = Path(os.path.dirname(__file__)) class OutData(BaseModel): name: str value: float @app.post("/multipart") async def run_multipart(input_data: str): out_data = OutData(name="A Name", value=10) shutil.make_archive(str(CUR_DIR), "zip", str(CUR_DIR)) zip_file = CUR_DIR.parent.joinpath(f"{CUR_DIR.name}.zip") # 构建Multipart编码器,拆分结构化数据和文件 encoder = MultipartEncoder( fields={ "out_data": ("out_data.json", json.dumps(out_data.model_dump()), "application/json"), "zip_file": ("archive.zip", open(zip_file, "rb"), "application/zip") } ) return StreamingResponse( iter([encoder.read()]), media_type=encoder.content_type )
外部requests调用示例
import requests import json url = "http://127.0.0.1:8000/multipart" response = requests.post(url, json={"input_data": "some"}) # 遍历Multipart响应的各个部分,分别提取数据和文件 for part in response.iter_multipart(): content_disposition = part.headers.get("Content-Disposition", "") if "name=\"out_data\"" in content_disposition: out_data = json.loads(part.text) elif "name=\"zip_file\"" in content_disposition: with open("downloaded_multipart.zip", "wb") as f: f.write(part.content) print(out_data)
注意事项
- 方案一中响应头有长度限制,若
out_data数据过大,优先选用方案二 - 生成的临时zip文件建议在接口返回后删除(可通过
finally块或临时文件模块实现),避免磁盘占用
内容的提问来源于stack exchange,提问作者aura
相关产品推荐
相关产品推荐

