FastAPIでストリーミングを使ってみる

import asyncio

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()


@app.get("/stream-json")
async def stream_json_example():
    async def generate_json_data():
        for i in range(10):
            data = {"index": i, "message": f"Hello from index {i}"}
            await asyncio.sleep(1.0)
            yield f"data:{json.dumps(data)}\n\n"


    return StreamingResponse(content=generate_json_data(), media_type="text/event-stream")