Last active
March 10, 2026 17:04
-
-
Save bbelderbos/06b783383b2c0aa1deb93e04dcf951d8 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import asyncio | |
| import random | |
| from datastar_py import ServerSentEventGenerator as SSE | |
| from datastar_py.fastapi import DatastarResponse | |
| from fastapi import FastAPI, Request | |
| from fastapi.responses import HTMLResponse | |
| from fastapi.templating import Jinja2Templates | |
| app = FastAPI(title="Datastar Stock Ticker") | |
| templates = Jinja2Templates(directory="templates") | |
| STOCKS = { | |
| "AAPL": {"name": "Apple", "price": 178.50}, | |
| "GOOGL": {"name": "Google", "price": 141.25}, | |
| "MSFT": {"name": "Microsoft", "price": 378.90}, | |
| "AMZN": {"name": "Amazon", "price": 178.75}, | |
| } | |
| def simulate_price_change(price: float) -> tuple[float, float]: | |
| """Simulate a small price change. | |
| Args: | |
| price: Current price (e.g., 100.0) | |
| Returns: | |
| Tuple of (new_price, change_percent) | |
| - new_price: price adjusted by -2% to +2% | |
| - change_percent: the % change applied | |
| Example: | |
| >>> simulate_price_change(100.0) | |
| (101.25, 1.25) # price went up 1.25% | |
| """ | |
| random_num = random.uniform(-2, 2) | |
| new_price = price * (1 + random_num / 100) | |
| return round(new_price, 2), round(random_num, 2) | |
| def render_ticker() -> str: | |
| """Render the stock ticker table body as HTML. | |
| CRITICAL: The element MUST have id="ticker" - this is how | |
| Datastar knows which DOM element to replace! | |
| Returns HTML like: | |
| <tbody id="ticker"> | |
| <tr> | |
| <td class="symbol">AAPL</td> | |
| <td class="name">Apple</td> | |
| <td class="price">$178.50</td> | |
| <td class="change green">▲ 1.25%</td> | |
| </tr> | |
| ...more rows... | |
| </tbody> | |
| """ | |
| rows = [] | |
| for symbol, data in STOCKS.items(): | |
| new_price, change = simulate_price_change(float(data["price"])) | |
| rows.append((symbol, data["name"], new_price, change)) | |
| table_rows = [] | |
| for row in sorted(rows, key=lambda x: x[2]): | |
| symbol, name, price, change = row | |
| arrow = "▲" if change > 0 else "▼" | |
| style = "green" if change > 0 else "red" | |
| tr = f""" | |
| <tr> | |
| <td class="symbol">{symbol}</td> | |
| <td class="name">{name}</td> | |
| <td class="price">$ {price}</td> | |
| <td class="change {style}">{arrow} {change}%</td> | |
| </tr> | |
| """ | |
| table_rows.append(tr) | |
| return f""" | |
| <tbody id="ticker"> | |
| {"\n".join(table_rows)} | |
| </tbody> | |
| """ | |
| @app.get("/", response_class=HTMLResponse) | |
| async def index(request: Request) -> HTMLResponse: | |
| """Serve the main page.""" | |
| return templates.TemplateResponse(request, "index.html") | |
| @app.get("/stream") | |
| async def stream(request: Request) -> DatastarResponse: | |
| """Stream stock price updates via SSE. | |
| This is the magic of Datastar: | |
| 1. Browser connects to /stream (triggered by data-init) | |
| 2. Server yields initial HTML immediately | |
| 3. Server loops forever: | |
| - Sleep 1 second | |
| - Update prices | |
| - Yield new HTML | |
| 4. Browser receives updates without refreshing! | |
| """ | |
| async def events(): | |
| # generators 😍 | |
| yield SSE.patch_elements(render_ticker()) | |
| while True: | |
| if await request.is_disconnected(): | |
| break | |
| await asyncio.sleep(1) | |
| yield SSE.patch_elements(render_ticker()) | |
| return DatastarResponse(events()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment