OpenAI SSE demo (#103)

Author: samuelcolvin

demo/__init__.py

@@ -14,6 +14,7 @@
 from .db import create_db
 from .forms import router as forms_router
 from .main import router as main_router
+from .sse import router as sse_router
 from .tables import router as table_router
 
 
@@ -33,6 +34,7 @@ async def lifespan(app_: FastAPI):
     app = FastAPI(lifespan=lifespan)
 
 app.include_router(components_router, prefix='/api/components')
+app.include_router(sse_router, prefix='/api/components')
 app.include_router(table_router, prefix='/api/table')
 app.include_router(forms_router, prefix='/api/forms')
 app.include_router(auth_router, prefix='/api/auth')

demo/components_list.py

@@ -1,11 +1,8 @@
 from __future__ import annotations as _annotations
 
 import asyncio
-from datetime import datetime
-from typing import AsyncIterable
 
 from fastapi import APIRouter
-from fastapi.responses import StreamingResponse
 from fastui import AnyComponent, FastUI
 from fastui import components as c
 from fastui.events import GoToEvent, PageEvent
@@ -144,7 +141,12 @@ class Delivery(BaseModel):
         c.Div(
             components=[
                 c.Heading(text='Server Load SSE', level=2),
-                c.Markdown(text='`ServerLoad` can also be used to load content from an SSE stream.'),
+                c.Markdown(
+                    text=(
+                        '`ServerLoad` can also be used to load content from an SSE stream.\n\n'
+                        "Here the response is the streamed output from OpenAI's GPT-4 chat model."
+                    )
+                ),
                 c.Button(text='Load SSE content', on_click=PageEvent(name='server-load-sse')),
                 c.Div(
                     components=[
@@ -219,21 +221,3 @@ class Delivery(BaseModel):
 async def modal_view() -> list[AnyComponent]:
     await asyncio.sleep(0.5)
     return [c.Paragraph(text='This is some dynamic content. Open devtools to see me being fetched from the server.')]
-
-
-async def sse_generator() -> AsyncIterable[str]:
-    while True:
-        d = datetime.now()
-        m = FastUI(
-            root=[
-                c.Div(components=[c.Text(text=f'Time {d:%H:%M:%S}')], class_name='font-monospace'),
-                c.Paragraph(text='This content is updated every second using an SSE stream.'),
-            ]
-        )
-        yield f'data: {m.model_dump_json(by_alias=True)}\n\n'
-        await asyncio.sleep(1)
-
-
-@router.get('/sse')
-async def sse_experiment() -> StreamingResponse:
-    return StreamingResponse(sse_generator(), media_type='text/event-stream')