英文:
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_all
和append_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")
请注意,我将帧拆分为first
和following
,以便第一个帧不会使其持续时间加倍(因为你隐式地包括了它两次)。其次,你的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.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论