python 从FastAPI返回多个文件

cuxqih21  于 2024-01-05  发布在  Python
关注(0)|答案(3)|浏览(408)

使用fastapi,我不知道如何发送多个文件作为响应。

  1. from fastapi import FastAPI, Response
  2. app = FastAPI()
  3. @app.get("/image_from_id/")
  4. async def image_from_id(image_id: int):
  5. # Get image from the database
  6. img = ...
  7. return Response(content=img, media_type="application/png")

字符串
然而,我不确定发送图像列表是什么样子的。理想情况下,我想这样做:

  1. @app.get("/images_from_ids/")
  2. async def image_from_id(image_ids: List[int]):
  3. # Get a list of images from the database
  4. images = ...
  5. return Response(content=images, media_type="multipart/form-data")


但是,这将返回错误

  1. def render(self, content: typing.Any) -> bytes:
  2. if content is None:
  3. return b""
  4. if isinstance(content, bytes):
  5. return content
  6. > return content.encode(self.charset)
  7. E AttributeError: 'list' object has no attribute 'encode'

oipij1gg

oipij1gg1#

我在Python 3和最新的fastapi上对@kia的答案有一些问题,所以这里是我得到的一个修复,它包括BytesIO而不是Stringio,响应属性的修复和顶级归档文件夹的删除

  1. import os
  2. import zipfile
  3. import io
  4. def zipfiles(filenames):
  5. zip_filename = "archive.zip"
  6. s = io.BytesIO()
  7. zf = zipfile.ZipFile(s, "w")
  8. for fpath in filenames:
  9. # Calculate path for file in zip
  10. fdir, fname = os.path.split(fpath)
  11. # Add file, at correct path
  12. zf.write(fpath, fname)
  13. # Must close zip for all contents to be written
  14. zf.close()
  15. # Grab ZIP file from in-memory, make response with correct MIME-type
  16. resp = Response(s.getvalue(), media_type="application/x-zip-compressed", headers={
  17. 'Content-Disposition': f'attachment;filename={zip_filename}'
  18. })
  19. return resp
  20. @app.get("/image_from_id/")
  21. async def image_from_id(image_id: int):
  22. # Get image from the database
  23. img = ...
  24. return zipfiles(img)

字符串

展开查看全部
mwg9r5ms

mwg9r5ms2#

此外,您可以即时创建zip,并使用StreamingResponse对象将其流回给用户:

  1. import os
  2. import zipfile
  3. import io
  4. from fastapi.responses import StreamingResponse
  5. zip_subdir = "/some_local_path/of_files_to_compress"
  6. def zipfile(filenames):
  7. zip_io = io.BytesIO()
  8. with zipfile.ZipFile(zip_io, mode='w', compression=zipfile.ZIP_DEFLATED) as temp_zip:
  9. for fpath in filenames:
  10. # Calculate path for file in zip
  11. fdir, fname = os.path.split(fpath)
  12. zip_path = os.path.join(zip_subdir, fname)
  13. # Add file, at correct path
  14. temp_zip.write(fpath, zip_path)
  15. return StreamingResponse(
  16. iter([zip_io.getvalue()]),
  17. media_type="application/x-zip-compressed",
  18. headers = { "Content-Disposition": f"attachment; filename=images.zip"}
  19. )

字符串

展开查看全部
jdg4fx2g

jdg4fx2g3#

压缩是最好的选择,将有相同的结果在所有浏览器。你可以压缩文件动态。

  1. import os
  2. import zipfile
  3. import StringIO
  4. def zipfiles(filenames):
  5. zip_subdir = "archive"
  6. zip_filename = "%s.zip" % zip_subdir
  7. # Open StringIO to grab in-memory ZIP contents
  8. s = StringIO.StringIO()
  9. # The zip compressor
  10. zf = zipfile.ZipFile(s, "w")
  11. for fpath in filenames:
  12. # Calculate path for file in zip
  13. fdir, fname = os.path.split(fpath)
  14. zip_path = os.path.join(zip_subdir, fname)
  15. # Add file, at correct path
  16. zf.write(fpath, zip_path)
  17. # Must close zip for all contents to be written
  18. zf.close()
  19. # Grab ZIP file from in-memory, make response with correct MIME-type
  20. resp = Response(s.getvalue(), mimetype = "application/x-zip-compressed")
  21. # ..and correct content-disposition
  22. resp['Content-Disposition'] = 'attachment; filename=%s' % zip_filename
  23. return resp
  24. @app.get("/image_from_id/")
  25. async def image_from_id(image_id: int):
  26. # Get image from the database
  27. img = ...
  28. return zipfiles(img)

字符串
作为替代,你可以使用base64编码来嵌入一个(非常小的)图像到json响应中。但我不推荐这样做。
你也可以使用MIME/multipart,但请记住,我是为电子邮件和/或POST传输到HTTP服务器而创建的。它从来没有打算在HTTP事务的客户端接收和解析。一些浏览器支持它,另一些不支持。(所以我认为你也不应该使用它)

展开查看全部

相关问题