데이터 스트리밍

데이터 스트리밍 (Stream Data)

응답을 한 번에 몽땅 주는 대신, 조각조각 스트리밍으로 보내고 싶은 때가 있어요. AI LLM 서비스의 출력처럼 문자열을 실시간으로 내보내거나, 큰 바이너리 파일을 전부 메모리에 올리지 않고 읽는 대로 보내고 싶을 때 말이죠. 이번 장에서는 그런 데이터 스트리밍을 FastAPI에서 어떻게 하는지 배워볼게요.

JSON으로 구조화할 수 있는 데이터를 스트리밍한다면 JSON Stream Lines 문서를 먼저 보세요. 여기서는 순수 바이너리 데이터나 문자열을 스트리밍하는 방법을 다룰게요.

참고 — 이 기능은 FastAPI 0.134.0에서 추가되었어요.

출처: 공식문서

어떤 경우에 쓰나요

순수 문자열을 스트리밍하고 싶을 때, 예를 들어 AI LLM 서비스의 출력을 그대로 내보낼 때.
큰 바이너리 파일을 전부 한 번에 메모리에 읽지 않고, 읽어 가는 각 덩어리(chunk)를 그대로 스트리밍할 때.
비디오나 오디오를 이렇게 스트리밍할 수 있어요. 처리하면서 보내는 것도 가능하고요.

`StreamingResponse`와 `yield`

경로 연산 함수에서 response_class=StreamingResponse를 선언하면, yield로 각 데이터 덩어리를 차례로 보낼 수 있어요.

from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


message = """
Rick: (stumbles in drunkenly, and turns on the lights) Morty! You gotta come on. You got--... you gotta come with me.
Morty: (rubs his eyes) What, Rick? What's going on?
Rick: I got a surprise for you, Morty.
Morty: It's the middle of the night. What are you talking about?
Rick: (spills alcohol on Morty's bed) Come on, I got a surprise for you. (drags Morty by the ankle) Come on, hurry up. (pulls Morty out of his bed and into the hall)
Morty: Ow! Ow! You're tugging me too hard!
Rick: We gotta go, gotta get outta here, come on. Got a surprise for you Morty.
"""


@app.get("/story/stream", response_class=StreamingResponse)
async def stream_story() -> AsyncIterable[str]:
    for line in message.splitlines():
        yield line


@app.get("/story/stream-no-async", response_class=StreamingResponse)
def stream_story_no_async() -> Iterable[str]:
    for line in message.splitlines():
        yield line


@app.get("/story/stream-no-annotation", response_class=StreamingResponse)
async def stream_story_no_annotation():
    for line in message.splitlines():
        yield line


@app.get("/story/stream-no-async-no-annotation", response_class=StreamingResponse)
def stream_story_no_async_no_annotation():
    for line in message.splitlines():
        yield line


@app.get("/story/stream-bytes", response_class=StreamingResponse)
async def stream_story_bytes() -> AsyncIterable[bytes]:
    for line in message.splitlines():
        yield line.encode("utf-8")


@app.get("/story/stream-no-async-bytes", response_class=StreamingResponse)
def stream_story_no_async_bytes() -> Iterable[bytes]:
    for line in message.splitlines():
        yield line.encode("utf-8")


@app.get("/story/stream-no-annotation-bytes", response_class=StreamingResponse)
async def stream_story_no_annotation_bytes():
    for line in message.splitlines():
        yield line.encode("utf-8")


@app.get("/story/stream-no-async-no-annotation-bytes", response_class=StreamingResponse)
def stream_story_no_async_no_annotation_bytes():
    for line in message.splitlines():
        yield line.encode("utf-8")

FastAPI는 각 데이터 덩어리를 StreamingResponse에 그대로 넘겨줘요. JSON으로 변환하거나 뭔가 비슷한 시도를 하지 않아요.

비동기가 아닌 경로 연산 함수

async 없는 일반 def 함수에서도 같은 방식으로 yield를 쓸 수 있어요. 위 코드의 stream_story_no_async가 그 예시예요.

애너테이션 없이

바이너리 데이터 스트리밍에서는 반환 타입 애너테이션을 굳이 선언하지 않아도 돼요. FastAPI가 그 데이터를 JSON으로 변환하거나 어떤 식으로든 직렬화하지 않으니까, 이 경우 타입 애너테이션은 에디터와 도구가 쓰는 용도일 뿐 FastAPI가 쓰지 않아요.

@app.get("/story/stream-no-annotation", response_class=StreamingResponse)
async def stream_story_no_annotation():
    for line in message.splitlines():
        yield line

이 말은 곧, StreamingResponse에서는 보내야 할 데이터 바이트를 타입 애너테이션과 무관하게 정확히 직접 만들고 인코딩할 자유(와 책임)가 있다는 뜻이에요. 🤓

바이트 스트리밍

주요 사용 사례 중 하나는 문자열 대신 bytes를 스트리밍하는 거예요. 당연히 할 수 있어요. stream_story_bytes처럼 yield line.encode("utf-8") 형태로 주면 돼요.

커스텀 `PNGStreamingResponse` 만들기

위 예제들에서는 데이터 바이트를 스트리밍했지만, 응답에 Content-Type 헤더가 없어서 클라이언트가 어떤 타입의 데이터를 받는지 몰랐어요.

StreamingResponse를 상속받아서, 스트리밍하는 데이터의 타입으로 Content-Type 헤더를 설정하는 커스텀 하위 클래스를 만들 수 있어요.

예를 들어 media_type 속성을 써서 Content-Type 헤더를 image/png로 설정하는 PNGStreamingResponse를 만들어 볼게요:

import base64
from collections.abc import AsyncIterable, Iterable
from io import BytesIO

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

image_base64 = "iVBORw0KGgoAAAANSUhEUgAAAB0AAAAdCAYAAABWk2cPAAAAbnpUWHRSYXcgcHJvZmlsZSB0eXBlIGV4aWYAAHjadYzRDYAwCET/mcIRDoq0jGOiJm7g+NJK0vjhS4DjIEfHfZ20DKqSrrWZmyFQV5ctRMOLACxglNCcXk7zVqFzJzF8kV6R5vOJ97yVH78HjfYAtg0ged033ZgAAAoCaVRYdFhNTDpjb20uYWRvYmUueG1wAAAAAAA8P3hwYWNrZXQgYmVnaW49Iu+7vyIgaWQ9Ilc1TTBNcENlaGlIenJlU3pOVGN6a2M5ZCI/Pgo8eDp4bXBtZXRhIHhtbG5zOng9ImFkb2JlOm5zOm1ldGEvIiB4OnhtcHRrPSJYTVAgQ29yZSA0LjQuMC1FeGl2MiI+CiA8cmRmOlJERiB4bWxuczpyZGY9Imh0dHA6Ly93d3cudzMub3JnLzE5OTkvMDIvMjItcmRmLXN5bnRheC1ucyMiPgogIDxyZGY6RGVzY3JpcHRpb24gcmRmOmFib3V0PSIiCiAgICB4bWxuczpleGlmPSJodHRwOi8vbnMuYWRvYmUuY29tL2V4aWYvMS4wLyIKICAgIHhtbG5zOnRpZmY9Imh0dHA6Ly9ucy5hZG9iZS5jb20vdGlmZi8xLjAvIgogICBleGlmOlBpeGVsWERpbWVuc2lvbj0iMjkiCiAgIGV4aWY6UGl4ZWxZRGltZW5zaW9uPSIyOSIKICAgdGlmZjpJbWFnZVdpZHRoPSIyOSIKICAgdGlmZjpJbWFnZUxlbmd0aD0iMjkiCiAgIHRpZmY6T3JpZW50YXRpb249IjEiLz4KIDwvcmRmOlJERj4KPC94OnhtcG1ldGE+CiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAKPD94cGFja2V0IGVuZD0idyI/PnQkBZAAAAAEc0JJVAgICAh8CGSIAAABoklEQVRIx8VXwY7FIAjE5iXWU+P/f6RHPNW9LIaOoHYP+0yMShVkwNGG1lqjfy4HfaF0oyEEt+oSQqBaa//m9Wd6PlqhhbRMDiEQM3e59FNKw5qZHpnQfuPaW6lazsztvu/eElFj5j63lNLlMz2ttbZtVMu1MTGo5Sujn93gMzOllKiUQjHGB9QxxneZhJ5iwZ1rL2fwenoGeL0q3wVGhBPHMz0PeFccIfASEeWcO8xEROd50q6eAV6s1s5XXoncas1EKqVQznnwUBdJJmm1l3hmmdlOMrGO8Vl5gZ56Y0y8IZF0BuqkQWM4B6HXrRCKa1SEqyzEo7KK59RT/VHDjX3ZvSefeW3CO6O6vsiA1NrwVkxxAcYTCcHyTjZmJd00pugBQoTnzjvn+kzLBh9GtRDjhleZFwbx3kugP3GvFzdkqRlbDYw0u/HxKjuOw2QxZCGL5V5f4l7cd6qsffUa1DcLM9N1XcTMvep5ul1e4jNPtZfWGIkE6dI8MquXg/dS2CGVJQ2ushd5GmlxFdOw+1tRa32MY4zDQ9yaZ60J3/iX+QG4U3qGrFHmswAAAABJRU5ErkJggg=="
binary_image = base64.b64decode(image_base64)


def read_image() -> BytesIO:
    return BytesIO(binary_image)


app = FastAPI()


class PNGStreamingResponse(StreamingResponse):
    media_type = "image/png"


@app.get("/image/stream", response_class=PNGStreamingResponse)
async def stream_image() -> AsyncIterable[bytes]:
    with read_image() as image_file:
        for chunk in image_file:
            yield chunk


@app.get("/image/stream-no-async", response_class=PNGStreamingResponse)
def stream_image_no_async() -> Iterable[bytes]:
    with read_image() as image_file:
        for chunk in image_file:
            yield chunk


@app.get("/image/stream-no-async-yield-from", response_class=PNGStreamingResponse)
def stream_image_no_async_yield_from() -> Iterable[bytes]:
    with read_image() as image_file:
        yield from image_file


@app.get("/image/stream-no-annotation", response_class=PNGStreamingResponse)
async def stream_image_no_annotation():
    with read_image() as image_file:
        for chunk in image_file:
            yield chunk


@app.get("/image/stream-no-async-no-annotation", response_class=PNGStreamingResponse)
def stream_image_no_async_no_annotation():
    with read_image() as image_file:
        for chunk in image_file:
            yield chunk

그리고 이 새 클래스를 경로 연산 함수의 response_class=PNGStreamingResponse로 쓰면 돼요. 위 코드의 /image/stream 경로들이 그 예시예요.

파일 흉내 내기 (Simulate a File)

이 예제에서는 io.BytesIO로 파일을 흉내 내고 있어요. io.BytesIO는 메모리에만 존재하는 파일 같은 객체(file-like object)인데, 인터페이스는 파일과 같아요. 예를 들어 파일처럼 반복해서 내용을 소비할 수 있어요.

with 블록을 쓰면 제너레이터 함수(yield가 있는 함수)가 끝난 뒤, 즉 응답 전송이 완료된 후에 파일 같은 객체가 닫히는 걸 보장해요. 이 예제에서는 가짜 메모리 파일(io.BytesIO)이라 크게 중요하지 않지만, 진짜 파일이라면 작업이 끝난 후 파일이 닫히는지 확인하는 게 중요해요.

참고 | 기술적 세부사항 — image_base64와 binary_image 변수는 이미지를 Base64로 인코딩한 다음 바이트로 바꾼 값이에요. 그래서 io.BytesIO로 넘길 수 있죠. 이 예제가 같은 파일 안에서 그대로 복사해 실행될 수 있도록 한 장치일 뿐이에요. 🥚

파일과 비동기 (Files and Async)

대부분의 경우 파일 같은 객체는 기본적으로 async/await와 호환되지 않아요. 예를 들어 await file.read()나 async for chunk in file 같은 걸 쓸 수 없어요.

그리고 대부분의 경우 파일을 읽는 건 블로킹(blocking) 작업이에요 (디스크나 네트워크에서 읽으니까). 그러면 이벤트 루프를 막을 수 있어요.

참고 — 위 예제는 예외에요. io.BytesIO 객체는 이미 메모리에 있으니 읽어도 아무것도 막지 않으니까요. 하지만 많은 경우 파일이나 파일 같은 객체를 읽으면 블로킹돼요.

이벤트 루프가 막히는 걸 피하려면 경로 연산 함수를 async def 대신 일반 def로 선언하면 돼요. 그러면 FastAPI가 그 함수를 **스레드풀 워커(threadpool worker)**에서 실행해서 메인 루프가 막히지 않게 해 줘요. 위 코드의 stream_image_no_async가 그 예시예요.

팁 — async 함수 안에서 블로킹 코드를 호출해야 하거나, 블로킹 함수 안에서 async 함수를 호출해야 한다면, FastAPI의 자매 라이브러리인 Asyncer를 쓸 수 있어요.

`yield from`

파일 같은 객체를 반복하면서 각 항목마다 yield를 하는 대신, yield from을 쓰면 각 항목을 직접 내보내고 for 루프를 생략할 수 있어요.

이건 FastAPI만의 얘기가 아니라 그냥 파이썬의 테크닉이지만, 알아두면 좋은 꿀팁이에요. 😎 위 코드의 stream_image_no_async_yield_from처럼 쓰면 돼요.

더 알아보기 (Learn more)

원문: Stream Data - FastAPI

데이터 스트리밍