[ FastAPI ] Streaming large size file using StreamingResponse

서버로 부터 이미지나 동영상 파일을 다운로드 받을 때, base64로 encode된 PlainTextResponse를 사용해도 되지만, 

좀 더 나은 성능을 보장하며 Async방식으로 다운 받을 수 있는 StreamingResponse 사용 및 사용 시 주의 점에 대해 알아보고자 합니다.

https://fastapi.tiangolo.com/advanced/custom-response/#streamingresponse

Custom Response - HTML, Stream, File, others - FastAPI

 

서버는 보통 클라이언트에서 요청한 S3에 저장된 이미지를 가져오고, 해당 이미지를 StreamingResponse로 내려줍니다.

이 때, 작은 사이즈의 파일들은 별다른 처리 없이 boto3 library를 사용하여 S3에서 받은 result의 body를 StreamingResponse에 넣어주면 이미지 파일이 받아집니다.

async def download_image_file(
        request: Request, 
    ) -> StreamingResponse:
    try:
        async with S3Client() as s3_client:
            result = await s3_client.download_file(key=key)
    except:
        ...
    return StreamingResponse(
        result["Body"], media_type=settings.media_type
    )

 

하지만, 큰 사이즈의 이미지 파일은 위 형태로 반환 시 이미지 파일을 chunk size로 나눈 첫번 째 chunk만 포함됩니다.

이는 StreamingResponse의 내부 구현과 관련이 있는데, 위 코드에서는 return을 통해 반환되어 함수가 끝나기 때문에 아래 stream_response에서 더 이상 context가 유지되지 않기에 나머지 chunk가 전달되지 않습니다.

async def stream_response(self, send: Send) -> None:
        await send(
            {
                "type": "http.response.start",
                "status": self.status_code,
                "headers": self.raw_headers,
            }
        )
        async for chunk in self.body_iterator:
            if not isinstance(chunk, bytes):
                chunk = chunk.encode(self.charset)
            await send({"type": "http.response.body", "body": chunk, "more_body": True})

        await send({"type": "http.response.body", "body": b"", "more_body": False})

 

이에 FastAPI의 공식 문서에서 가이드 하듯, StreamingResponse의 content인자에 async generator를 전달해주어, context를 유지하여 yield와 반복문을 통해 chunk를 순회하며 streaming하는 방식으로 전달해야 합니다.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer())

이렇게 되면 return을 하더라도 StreamingResponse를 반환한 곳에서 context가 유지되고 stream_response함수에서 지속적으로 await send()를 통해 남은 chunk들을 순서대로 보낼 수 있게 됩니다. 아래는, 위 download_image_file함수를 변환한 것으로, S3에서 받은 큰  사이즈의 이미지를 StreamingResponse로 반환하는 예시입니다.

async def download_image_file(
        request: Request, 
    ) -> StreamingResponse:
    try:
        async def download_chunks() -> AsyncGenerator[bytes, None]:
            async with S3Client() as s3_client:
                result = await s3_client.download_file(key=key)
                async for chunk in result["Body"].iter_chunks():
                    yield chunk

        return StreamingResponse(
            content=download_chunks(), media_type=settings.media_type
        )
    except:
        ...

물론 서버에서 Streaming이 필요없다면, 메모리에 이미지를 저장하고 한번에 내려줄 수도 있지만 스트리밍을 통해 성능 상의 이점을 가지고 싶다면 위 방식대로 구현해보는 것도 좋을 것 같습니다. 감사합니다 😊🙌