Example of Datastar with FastHTML, ported from a FastAPI example

main.py

import asyncio
import json
from datetime import datetime

from fasthtml.common import *
from fasthtml.starlette import StreamingResponse
from fasthtml.core import to_xml
from datastar_py.sse import SSE_HEADERS, ServerSentEventGenerator


# Create a FastHTML response based on the datastar SDK, allowing FastTags to be used directly
class DatastarFastHTMLResponse(StreamingResponse):
    def __init__(self, generator, *args, **kwargs):
        kwargs["headers"] = SSE_HEADERS

        class XMLServerSentEventGenerator(ServerSentEventGenerator):
            @classmethod
            def merge_fragments(cls, fragments, *args, **kwargs):
                # Run to_xml to convert all fragments from FastTags to XML
                xml_fragments = [
                    to_xml(f) if hasattr(f, "render") else f for f in fragments
                ]
                # From here, business as usual
                return super().merge_fragments(xml_fragments, *args, **kwargs)

        super().__init__(generator(XMLServerSentEventGenerator), *args, **kwargs)


# Import datastar from a CDN and as a module
datastar_src = Script(
    type="module",
    src="https://cdn.jsdelivr.net/gh/starfederation/datastar@v1.0.0-beta.9/bundles/datastar.js",
)


app, rt = fast_app(live=True, hdrs=(datastar_src,))

example_style = Style(
    "html, body { height: 100%; width: 100%; } h1 { color: #ccc; text-align: center } body { background-image: linear-gradient(to right bottom, oklch(0.424958 0.052808 253.972015), oklch(0.189627 0.038744 264.832977)); } .container { display: grid; place-content: center; } .time { padding: 2rem; border-radius: 8px; margin-top: 3rem; font-family: monospace, sans-serif; background-color: oklch(0.916374 0.034554 90.5157); color: oklch(0.265104 0.006243 0.522862 / 0.6); font-weight: 600; }"
)


# Initial response returns HTML
@rt("/")
async def index():
    now = datetime.isoformat(datetime.now())
    return Titled(
        "Datastar FastHTML example",
        example_style,
        # Init currentTime signal with formatted time
        # 3 different ways of writing the same thing, the last one I like the best
        # Body(data_signals=f"{{'currentTime': '{now}'}}")(
        # Body({"data-signals": f"{{currentTime: '{now}'}}"})(
        Body(data_signals=json.dumps({"currentTime": now}))(
            Div(cls="container")(
                # Initiate a GET request on load to the /updates endpoint, which returns a stream
                Div(data_on_load="@get('/updates')", cls="time")(
                    "Current time from fragment: ",
                    # Use currentTime ID for fragment replacement
                    Span(id="currentTime")(now),
                ),
                Div(cls="time")(
                    "Current time from signal: ",
                    # Use currentTime as data-text for signal replacement
                    Span(data_text="$currentTime")(now),
                ),
                Div(cls="time")(
                    "Counted seconds: ",
                    # Initiate count of number of seconds so far counted, also using an ID
                    Span(id="myNumber")(0),
                ),
            )
        ),
    )


async def clock(sse):
    n = 0
    while True:
        # Constantly get current time
        now = datetime.isoformat(datetime.now())
        # Replace 2 DOM elements at the same time:
        # 1. The element with the ID 'currentTime' with the current time
        # 2. The element with the ID 'myNumber' with the current count n
        yield sse.merge_fragments([Span(id="currentTime")(now), Span(id="myNumber")(n)])
        n += 1
        # Wait 1 second
        await asyncio.sleep(1)
        # Replace 2 more things:
        # 1. The signal named 'currentTime' with the current time
        # This is automatically injected into the element with attribute data-text='currentTime'
        yield sse.merge_signals({"currentTime": f"{now}"})
        # 2. The element with the ID 'myNumber' with the current count n
        yield sse.merge_fragments([Span(id="myNumber")(n)])
        n += 1
        # Wait another second
        await asyncio.sleep(1)


@rt("/updates")
async def updates():
    # Use the above defined DatastarFastHTMLResponse with the generator function defined above as the response
    return DatastarFastHTMLResponse(clock)


serve()