Coverage for docs_src / server_sent_events / tutorial001_py310.py: 100%

1from collections.abc import AsyncIterable, Iterable 9 ctx1abhicdefg

2 

3from fastapi import FastAPI 9 ctx1abhicdefg

4from fastapi.sse import EventSourceResponse 9 ctx1abhicdefg

5from pydantic import BaseModel 9 ctx1abhicdefg

6 

7app = FastAPI() 9 ctx1abhicdefg

8 

9 

10class Item(BaseModel): 9 ctx1abhicdefg

11 name: str 7 ctx1abcdefg

12 description: str | None 7 ctx1abcdefg

13 

14 

15items = [ 9 ctx1abhicdefg

16 Item(name="Plumbus", description="A multi-purpose household device."), 

17 Item(name="Portal Gun", description="A portal opening device."), 

18 Item(name="Meeseeks Box", description="A box that summons a Meeseeks."), 

19] 

20 

21 

22@app.get("/items/stream", response_class=EventSourceResponse) 9 ctx1abhicdefg

23async def sse_items() -> AsyncIterable[Item]: 9 ctx1abhicdefg

24 for item in items: 4 ctx1jklm

25 yield item 4 ctx1jklm

26 

27 

28@app.get("/items/stream-no-async", response_class=EventSourceResponse) 9 ctx1abhicdefg

29def sse_items_no_async() -> Iterable[Item]: 9 ctx1abhicdefg

30 for item in items: 4 ctx1nopq

31 yield item 4 ctx1nopq

32 

33 

34@app.get("/items/stream-no-annotation", response_class=EventSourceResponse) 9 ctx1abhicdefg

35async def sse_items_no_annotation(): 9 ctx1abhicdefg

36 for item in items: 4 ctx1rstu

37 yield item 4 ctx1rstu

38 

39 

40@app.get("/items/stream-no-async-no-annotation", response_class=EventSourceResponse) 9 ctx1abhicdefg

41def sse_items_no_async_no_annotation(): 9 ctx1abhicdefg

42 for item in items: 4 ctx1vwxy

43 yield item 4 ctx1vwxy