axelknock · January 29, 2025 22:21 · Kvit · Jan 29, 2025
diff --git a/main.py b/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/[email protected]/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()
	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/[email protected]/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()
No results found