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."), ] @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