FastHTML Examples

This implementation enables a seamless CLI-to-Browser authentication flow using a paircode to link the two environments.

The Authentication Lifecycle

1. Paircode Generation: The CLI client generates a unique paircode (e.g., using secrets.token_urlsafe(16)) to identify the session.
2. Handshake: The client sends the paircode to the server's /cli_login endpoint. The server stores this code and returns a login URL containing the paircode as the state parameter.
3. Browser Flow: The client opens the login URL in the user's browser. The user authenticates via an OAuth provider, which then redirects back to the server's redirect endpoint with an authorization code and the original state (paircode).
4. Token Association: The server exchanges the code for an authentication ID and associates it with the paircode in its internal store (pc_store).
5. Polling & Retrieval: The CLI client polls the server's /token endpoint using the paircode. Once the server detects the completed authentication, it returns the session cookies.
6. Persistence: The client saves these cookies (e.g., to auth_token.txt) and attaches them to subsequent httpx.Client requests to access secured endpoints.

# Client generates unique identifier
paircode = secrets.token_urlsafe(16)

# Client polls for the token
def poll_token(paircode, host, interval=1, timeout=180):
    # ... implementation polls /token?paircode={paircode} ...
    return dict(client.cookies)

This project uses FastHTML's @app.ws decorator to manage WebSocket connections for multi-client synchronization.

Key patterns used:

• Connection Management: Use on_connect and on_disconnect callbacks to manage a queue of active client connections (player_queue).
• Out-of-Band (OOB) Swaps: Instead of standard HTMX requests returning HTML to the immediate caller, the server pushes FastHTML components to all connected clients via WebSockets. This ensures that an action taken by one user (e.g., clicking a cell) is reflected on all other users' screens.
• Background Updates: Use an asyncio background task to continuously evolve the game state and broadcast updates to all players in the player_queue using a function like update_players().

@app.ws('/gol', conn=on_connect, disconn=on_disconnect)
def ws(msg:str, send): pass

player_queue = []
async def on_connect(send): player_queue.append(send)
async def on_disconnect(send): await update_players()

To implement infinite scrolling, place a sentinel Div at the end of your content list. This Div uses the hx_trigger="intersect once" attribute to detect when it enters the viewport, triggering a request to fetch the next batch of data.

To optimize for slow-loading content, you can place this sentinel element a few items before the actual end of the list so the next batch begins loading before the user reaches the bottom.

Key HTMX attributes used:

• hx_get: The endpoint to fetch the next set of items.
• hx_trigger="intersect once": Triggers the request when the element becomes visible in the viewport, but only once per element.
• hx_swap="afterend": Appends the new content immediately after the sentinel element.
• hx_target="this": Targets the sentinel element itself for the swap operation.

@app.get("/more-cards")
def more_cards(request):
    # Get the current count from the query parameters
    start = int(request.query_params.get("start", 21))
    end = start + 20
    
    new_cards = [create_card(i) for i in range(start, end)]
    
    return *new_cards, Div(
            hx_get=f"/more-cards?start={end}",
            hx_trigger="intersect once",
            hx_swap="afterend",
            hx_target="this"
        )

To stream large responses using Transfer-Encoding: chunked, use the htmx-ext-transfer-encoding-chunked extension.

1. Setup: Pass exts='chunked-transfer' to the FastHTML constructor.
2. Form: Add hx_ext="chunked-transfer" to the form.
3. Server: Return a StreamingResponse from your handler. Use yield to send chunks of HTML (wrapped in to_xml) to the client.

Example Handler:

@app.post
async def send(msg:str, messages:list[str]=None):
    if not messages: messages = []
    messages.append(msg.rstrip())
    return StreamingResponse(stream_response(msg, messages), media_type="text/plain", headers={"X-Transfer-Encoding": "chunked"})

async def stream_response(msg, messages):
    # Initial setup and streaming loop
    yield to_xml(ChatMessage(msg, True, id=len(messages)-1))
    yield to_xml(ChatMessage('', False, id=len(messages)))
    r = (cli(messages, sp=sp, stream=True))
    response_txt = ''
    for chunk in r:
        response_txt += chunk
        yield to_xml(Div(
            response_txt,
            cls=f"chat-bubble chat-bubble-secondary",
            id=f"msg-{len(messages)}-content",
            hx_swap_oob="outerHTML",
        ))
        await asyncio.sleep(0.2)
    # ... (final cleanup yields)