JSON Lines로 데이터 스트리밍하기

JSON Lines로 데이터 스트리밍하기

클라이언트에 보내고 싶은 데이터 시퀀스가 끝까지 준비되기를 기다리지 않고, 만들어지는 대로 한 항목씩 보내고 싶을 때가 있어요. 이런 경우 JSON Lines 형식이 잘 맞아요. 각 줄에 JSON 객체 하나를 보내는 방식인데요, FastAPI의 path operation 함수에서 return 대신 yield를 쓰면 간단하게 구현할 수 있어요.

출처: FastAPI 공식 문서 - Stream JSON Lines

참고 — FastAPI 0.134.0에 추가된 기능이에요.

스트리밍이 뭔가요

스트리밍 데이터는 앱이 전체 항목 시퀀스가 준비될 때까지 기다리지 않고, 첫 항목부터 클라이언트로 보내기 시작한다는 뜻이에요. 첫 항목을 보내면 클라이언트는 그것부터 받아서 처리하기 시작하고, 그 사이에 앱은 다음 항목을 계속 만들어 낼 수 있어요. 심지어 계속해서 데이터를 보내는 무한 스트림도 가능해요.

sequenceDiagram
    participant App
    participant Client

    App->>App: Produce Item 1
    App->>Client: Send Item 1
    App->>App: Produce Item 2
    Client->>Client: Process Item 1
    App->>Client: Send Item 2
    App->>App: Produce Item 3
    Client->>Client: Process Item 2
    App->>Client: Send Item 3
    Client->>Client: Process Item 3
    Note over App: Keeps producing...
    Note over Client: Keeps consuming...

JSON Lines 형식

JSON Lines는 한 줄에 JSON 객체 하나를 보내는 형식이에요. 응답의 content type은 application/json이 아니라 application/jsonl이고, 본문은 이런 모양이에요.

{"name": "Plumbus", "description": "A multi-purpose household device."}
{"name": "Portal Gun", "description": "A portal opening device."}
{"name": "Meeseeks Box", "description": "A box that summons a Meeseeks."}

JSON 배열(파이썬의 리스트)과 비슷하지만 []로 감싸고 ,로 구분하는 대신, 한 줄에 객체 하나를 두고 줄바꿈 문자로 구분해요. 핵심은 앱이 각 줄을 차례로 만들어 내는 동안 클라이언트가 이전 줄을 소비한다는 점이에요.

기술적으로 각 JSON 객체가 개행으로 분리되기 때문에 내용에 리터럴 줄바꿈 문자를 넣을 수는 없지만, JSON 표준의 일부인 이스케이프된 줄바꿈(\n)은 포함할 수 있어요. 다만 보통은 자동으로 처리되니 걱정하지 않아도 돼요.

어떤 데 쓸까요

이 방식을 AI LLM 서비스에서 데이터를 스트리밍하거나, 로그·텔레메트리, 또는 JSON 항목으로 구조화할 수 있는 다른 종류의 데이터를 보낼 때 쓸 수 있어요. 바이너리 데이터(예: 동영상·오디오)를 스트리밍하려면 Stream Data 고급 가이드를 확인해 보세요.

FastAPI에서 JSON Lines 스트리밍하기

FastAPI에서 JSON Lines를 스트리밍하려면 path operation 함수에서 return 대신 yield를 써서 각 항목을 차례로 만들어 내면 돼요. 보낼 각 JSON 항목이 Item(Pydantic 모델) 타입이고 비동기 함수라면, 반환 타입을 AsyncIterable[Item]로 선언할 수 있어요.

from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-async")
def stream_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-annotation")
async def stream_items_no_annotation():
    for item in items:
        yield item


@app.get("/items/stream-no-async-no-annotation")
def stream_items_no_async_no_annotation():
    for item in items:
        yield item

반환 타입을 선언하면 FastAPI가 그 타입을 이용해 데이터를 검증하고, OpenAPI에 문서화하고, 필터링하고, Pydantic으로 직렬화해요. Pydantic이 직렬화를 Rust 쪽에서 처리하므로, 반환 타입을 선언하지 않을 때보다 훨씬 높은 성능을 얻을 수 있어요.

비동기가 아닌 path operation 함수

async가 없는 일반 def 함수에서도 똑같이 yield를 쓸 수 있어요. FastAPI가 이벤트 루프를 막지 않도록 올바르게 실행해 줘요. 이 경우 함수가 비동기가 아니므로 올바른 반환 타입은 Iterable[Item]이에요.

반환 타입 없이 쓰기

반환 타입을 생략할 수도 있어요. 그러면 FastAPI가 jsonable_encoder로 데이터를 JSON으로 직렬화할 수 있는 형태로 변환해서 JSON Lines로 보내요.

Server-Sent Events (SSE)로 넘어가기

FastAPI는 Server-Sent Events(SSE)도 일급 지원해요. JSON Lines와 비슷하지만 몇 가지 추가 세부 사항이 있는데요, 그 내용은 다음 챕터에서 다뤄요.

더 알아보기 (Learn more)