如何将内存中的动画 GIF 发送到 FastAPI 端点?

huangapple go评论77阅读模式
英文:

How to send animated GIF from memory to FastAPI endpoint?

问题

# main.py

@app.get('/youtube/{video_id}.gif')
def youtube_thumbnail(video_id: str, width: int = 320, height: int = 180):
    image = util.create_youtube_gif_thumbnail(video_id)
    return util.gif_to_streaming_response(image)
# util.py

def gif_to_streaming_response(image: Image):
    imgio = BytesIO()
    image.save(imgio, 'GIF')
    imgio.seek(0)
    return StreamingResponse(content=imgio, media_type="image/gif")


def create_gif(images, duration):
    gif = images[0]
    output = BytesIO() 
    output = "./file.gif"
    gif.save(output, format='GIF', save_all=True, append_images=images, loop=0, duration=duration)
    return gif


def create_youtube_gif_thumbnail(video_id: str):
    images = [read_img_from_url(f"https://i.ytimg.com/vi/{video_id}/{i}.jpg") for i in range(1,3)]
    return create_gif(images, 150)

Here's the full codebase: https://github.com/Snailedlt/Markdown-Videos/blob/418de75e200bf9d9f4f02e5a667af4c9b226b5d3/util.py#L74

英文:

I'm trying to send a GIF from memory to my FastAPI endpoint, which works, but the gif isn't animated. When I save it locally instead the animation works fine.
I don't want to save the image, but instead keep it in memory until it's returned to the endpoint.

I've already checked out this post, but still couldn't get it working: https://stackoverflow.com/questions/67571477/python-fastapi-returned-gif-image-is-not-animating

So how can I return an animated .gif using FastAPI?

This is my attempted solution:

# main.py

@app.get('/youtube/{video_id}.gif')
def youtube_thumbnail (video_id: str, width: int = 320, height: int = 180):
    image = util.create_youtube_gif_thumbnail(video_id)
    return util.gif_to_streaming_response(image)
# util.py

def gif_to_streaming_response(image: Image):
    imgio = BytesIO()
    image.save(imgio, 'GIF')
    imgio.seek(0)
    return StreamingResponse(content=imgio, media_type="image/gif")


def create_gif(images, duration):
    gif = images[0]
    output = BytesIO() # for some reason this doesn't work (it shows a still image), but if I save the image with output = "./file.gif" the animation works!
    # this works
    output = "./file.gif"
    gif.save(output, format='GIF', save_all=True, append_images=images, loop=0, duration=duration)
    return gif


def create_youtube_gif_thumbnail(video_id: str):
    images = [read_img_from_url(f"https://i.ytimg.com/vi/{video_id}/{i}.jpg") for i in range(1,3)]
    return create_gif(images, 150)

Here's the full codebase: https://github.com/Snailedlt/Markdown-Videos/blob/418de75e200bf9d9f4f02e5a667af4c9b226b5d3/util.py#L74

答案1

得分: 1

你的代码很难跟踪,但看起来你正在保存不同的东西到你的端点返回的内容。

如果我将大部分你的代码合并为一个单一的方法,清楚了解正在发生的事情:

@app.get('/youtube/{video_id}.gif')
def youtube_thumbnail(video_id: str, width: int = 320, height: int = 180):
    # from util.create_youtube_gif_thumbnail(video_id)
    images = [read_img_from_url(f"https://i.ytimg.com/vi/{video_id}/{i}.jpg") for i in range(1,3)]
    # from create_gif(images, 150)
    image = images[0]
    image.save("./file.gif", format='GIF', save_all=True, append_images=images, loop=0, duration=150)
    # from util.gif_to_streaming_response(image)
    imgio = BytesIO()
    image.save(imgio, 'GIF')
    imgio.seek(0)
    return StreamingResponse(content=imgio, media_type="image/gif")

因此,在写入到file.gif时,你传递了save_allappend_images,但没有作为StreamingResponse的一部分。

也就是说,它实际上是这样的:

@app.get('/youtube/{video_id}.gif')
def youtube_thumbnail(video_id: str):
    images = [read_img_from_url(f"https://i.ytimg.com/vi/{video_id}/{i}.jpg") for i in range(1,3)]
    image = images[0]
    imgio = BytesIO()
    image.save(imgio, 'GIF')
    imgio.seek(0)
    return StreamingResponse(content=imgio, media_type="image/gif")

但我建议这样做:

@app.get('/youtube/{video_id}.gif')
def youtube_thumbnail(video_id: str):
    first, *following = (
        read_img_from_url(f"https://i.ytimg.com/vi/{video_id}/{i+1}.jpg")
        for i in range(3)
    )
    buf = BytesIO()
    first.save(buf, format='GIF', save_all=True, append_images=following, loop=0, duration=150)
    return Response(buf.getvalue(), media_type="image/gif")

请注意,我将帧拆分为firstfollowing,以便第一个帧不会使其持续时间加倍(因为你隐式地包括了它两次)。其次,你的range(1,3)仅迭代[1, 2]而不是似乎是我尝试的视频的有效范围[1, 2, 3]。最后,使用StreamingResponse没有太多的价值,因为整个响应已经在内存中(由于使用BytesIO),因此直接将bytearray传递给Response似乎是可以接受的。

我还建议修改read_img_from_url以检查错误,类似于:

def read_img_from_url(url: str) -> Image.Image:
    res = requests.get(url)
    res.raise_for_status()
    return Image.open(BytesIO(res.content))

很烦人的是,requests不会自动引发这样的错误。你可能仍然需要使用try/catch块,因为解析器/套接字异常可能在内部引发。

英文:

Your code is very difficult to follow, but it looks like you're saving something different to what you're returning from your endpoint.

If I flatten most of your code into a single method it's clearer what's going on:

@app.get('/youtube/{video_id}.gif')
def youtube_thumbnail (video_id: str, width: int = 320, height: int = 180):
    # from util.create_youtube_gif_thumbnail(video_id)
    images = [read_img_from_url(f"https://i.ytimg.com/vi/{video_id}/{i}.jpg") for i in range(1,3)]
    # from create_gif(images, 150)
    image = images[0]
    image.save("./file.gif", format='GIF', save_all=True, append_images=images, loop=0, duration=150)
    # from util.gif_to_streaming_response(image)
    imgio = BytesIO()
    image.save(imgio, 'GIF')
    imgio.seek(0)
    return StreamingResponse(content=imgio, media_type="image/gif")

So you're passing save_all and append_images when writing to file.gif, but not as part of the StreamingResponse.

I.e., it's effectively doing:

@app.get('/youtube/{video_id}.gif')
def youtube_thumbnail (video_id: str):
    images = [read_img_from_url(f"https://i.ytimg.com/vi/{video_id}/{i}.jpg") for i in range(1,3)]
    image = images[0]
    imgio = BytesIO()
    image.save(imgio, 'GIF')
    imgio.seek(0)
    return StreamingResponse(content=imgio, media_type="image/gif")

While, I'd suggest doing:

@app.get('/youtube/{video_id}.gif')
def youtube_thumbnail (video_id: str):
    first, *following = (
        read_img_from_url(f"https://i.ytimg.com/vi/{video_id}/{i+1}.jpg")
        for i in range(3)
    )
    buf = BytesIO()
    first.save(buf, format='GIF', save_all=True, append_images=following, loop=0, duration=150)
    return Response(buf.getvalue(), media_type="image/gif")

Note that I've split the frames up into first and following so that the first doesn't have its duration doubled (as you were implicitly including it twice). Next, your range(1,3) only iterates over [1, 2] rather than [1, 2, 3] which seem to be the valid range for the videos I tried. Finally, there is little value in using a StreamingResponse as the whole response is already in memory (due to the use of BytesIO) hence just passing the bytearray directly to a Response seems acceptable.

I'd also suggest modifying read_img_from_url to check for errors, something like:

def read_img_from_url(url: str) -> Image.Image:
    res = requests.get(url)
    res.raise_for_status()
    return Image.open(BytesIO(res.content))

It's annoying that requests doesn't raise errors like this automatically. You likely need a try/catch block anyway as resolver/socket exceptions might be raised internally.

huangapple
  • 本文由 发表于 2023年6月29日 07:30:34
  • 转载请务必保留本文链接:https://go.coder-hub.com/76577266.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定