如何 return fastAPI 中的图像?

How do I return an image in fastAPI?

使用 python 模块 fastAPI,我不知道如何 return 图像。在烧瓶中我会做这样的事情:

@app.route("/vector_image", methods=["POST"])
def image_endpoint():
    # img = ... # Create the image here
    return Response(img, mimetype="image/png")

这个模块对应的调用是什么?

尚未正确记录,但您可以使用 Starlette 的任何内容。

因此,如果它是磁盘中路径为 https://www.starlette.io/responses/#fileresponse

的文件,您可以使用 FileResponse

如果它是在您的 路径操作中创建的类文件对象,在 Starlette 的下一个稳定版本(由 FastAPI 内部使用)中,您还可以 return 它在 StreamingResponse.

@SebastiánRamírez 的 为我指明了正确的方向,但对于那些希望解决问题的人来说,我需要几行代码才能使其正常工作。我需要从 starlette(不是 fastAPI?)导入 FileResponse,添加 CORS 支持,并从临时文件中导入 return。也许有更好的方法,但我无法使用流式传输:

from starlette.responses import FileResponse
from starlette.middleware.cors import CORSMiddleware
import tempfile

app = FastAPI()
app.add_middleware(
    CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]
)

@app.post("/vector_image")
def image_endpoint(*, vector):
    # Returns a raw PNG from the document vector (define here)
    img = my_function(vector)

    with tempfile.NamedTemporaryFile(mode="w+b", suffix=".png", delete=False) as FOUT:
        FOUT.write(img)
        return FileResponse(FOUT.name, media_type="image/png")

我遇到了类似的问题,但使用的是 cv2 图像。这可能对其他人有用。使用 StreamingResponse.

import io
from starlette.responses import StreamingResponse

app = FastAPI()

@app.post("/vector_image")
def image_endpoint(*, vector):
    # Returns a cv2 image array from the document vector
    cv2img = my_function(vector)
    res, im_png = cv2.imencode(".png", cv2img)
    return StreamingResponse(io.BytesIO(im_png.tobytes()), media_type="image/png")

感谢@biophetik 的回答,有一个重要的提醒让我感到困惑:如果您正在使用 BytesIO,尤其是 PIL/skimage,请务必也执行 img.seek(0)回来之前!

@app.get("/generate")
def generate(data: str):
  img = generate_image(data)
  print('img=%s' % (img.shape,))
  buf = BytesIO()
  imsave(buf, img, format='JPEG', quality=100)
  buf.seek(0) # important here!
  return StreamingResponse(buf, media_type="image/jpeg",
    headers={'Content-Disposition': 'inline; filename="%s.jpg"' %(data,)})

所有其他答案都正确,但现在 return 一张图片

变得如此简单
from fastapi.responses import FileResponse

@app.get("/")
async def main():
    return FileResponse("your_image.jpeg")

您可以在 FastAPI 中做一些非常相似的事情

from fastapi import FastAPI, Response

app = FastAPI()

@app.post("/vector_image/")
async def image_endpoint():
    # img = ... # Create the image here
    return Response(content=img, media_type="image/png")

如果内存中已经有图像的字节

Return 一个 fastapi.responses.Response 与您的自定义 contentmedia_type.

您还需要处理端点装饰器以使 FastAPI 将正确的媒体类型放入 OpenAPI 规范中。

@app.get(
    "/image",

    # Set what the media type will be in the autogenerated OpenAPI specification.
    # fastapi.tiangolo.com/advanced/additional-responses/#additional-media-types-for-the-main-response
    responses = {
        200: {
            "content": {"image/png": {}}
        }
    }

    # Prevent FastAPI from adding "application/json" as an additional
    # response media type in the autogenerated OpenAPI specification.
    # https://github.com/tiangolo/fastapi/issues/3258
    response_class=Response,
)
def get_image()
    image_bytes: bytes = generate_cat_picture()
    # media_type here sets the media type of the actual response sent to the client.
    return Response(content=image_bytes, media_type="image/png")

参见Response documentation

如果您的图像仅存在于文件系统中

Return一个fastapi.responses.FileResponse.

参见FileResponse documentation


小心StreamingResponse

其他答案建议 StreamingResponseStreamingResponse 更难正确使用,所以我不推荐它,除非你确定你不能使用 ResponseFileResponse.

特别是,这样的代码毫无意义。它不会以任何有用的方式“流式传输”图像。

@app.get("/image")
def get_image()
    image_bytes: bytes = generate_cat_picture()
    # ❌ Don't do this.
    image_stream = io.BytesIO(image_bytes)
    return StreamingResponse(content=image_stream, media_type="image/png")

首先,StreamingResponse(content=my_iterable) 通过迭代 my_iterable 提供的数据块进行流式传输。但是当那个可迭代对象是 BytesIOthe chunks will be \n-terminated lines 时,这对二值图像没有意义。

即使分块有意义,分块在这里也毫无意义,因为我们从一开始就拥有整个 image_bytes bytes 对象。我们不妨从一开始就将整个事情传递给 Response。我们不会通过从 FastAPI 获取数据来获得任何好处。

其次,StreamingResponse 至少对应于 HTTP chunked transfer encoding. (This might depend on your ASGI server, but it's the case for Uvicorn。)这不是分块传输编码的好用例。

当您事先不知道输出的大小时,并且您不想在开始将其发送给客户端之前等到全部收集起来才知道时,分块传输编码是有意义的。这可以适用于诸如提供慢速数据库查询结果之类的东西,但它通常不适用于提供图像。

不必要的分块传输编码可能有害。例如,这意味着客户端在下载文件时无法显示进度条。参见:

  • Content-Length header versus chunked encoding

如果磁盘中的文件具有 path:

,您可以使用 FileResponse
import os

from fastapi import FastAPI 
from fastapi.responses import FileResponse

app = FastAPI()

path = "/path/to/files"

@app.get("/")
def index():
    return {"Hello": "World"}

@app.get("/vector_image", responses={200: {"description": "A picture of a vector image.", "content" : {"image/jpeg" : {"example" : "No example available. Just imagine a picture of a vector image."}}}})
def image_endpoint():
    file_path = os.path.join(path, "files/vector_image.jpg")
    if os.path.exists(file_path):
        return FileResponse(file_path, media_type="image/jpeg", filename="vector_image_for_you.jpg")
    return {"error" : "File not found!"}

由于我的镜像是用PIL构建的,所以上面的我的需求并没有完全满足。我的 fastapi 端点获取一个图像文件名,将其读取为 PIL 图像,并在内存中生成一个缩略图 jpeg,可以在 HTML 中使用,例如:

<img src="http://localhost:8000/images/thumbnail/bigimage.jpg">

import io
from PIL import Image
from fastapi.responses import StreamingResponse
@app.get('/images/thumbnail/{filename}',
  response_description="Returns a thumbnail image from a larger image",
  response_class="StreamingResponse",
  responses= {200: {"description": "an image", "content": {"image/jpeg": {}}}})
def thumbnail_image (filename: str):
  # read the high-res image file
  image = Image.open(filename)
  # create a thumbnail image
  image.thumbnail((100, 100))
  imgio = io.BytesIO()
  image.save(imgio, 'JPEG')
  imgio.seek(0)
  return StreamingResponse(content=imgio, media_type="image/jpeg")