PyCafe - Asgi - WebSocket Interaction with FastHTML

app.py
requirements.txt
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# Run with: python basic_app.py
from fasthtml.common import *

# TODO: this should happen automatically in the future
import jc.common
jc.common.patch_anyio()

from asyncio import sleep
from fasthtml.common import *

app = FastHTML(ws_hdr=True)
rt = app.route

def mk_inp(): return Input(id='msg')
nid = 'notifications'

@rt('/')
async def get():
    cts = Div(
        Div(id=nid),
        Form(mk_inp(), id='form', ws_send=True),
        # TODO: is there no other way than to prepend the ROOT_PATH?
        hx_ext='ws', ws_connect=jc.asgi.ROOT_PATH+'/ws')
    
    # TODO: this is not needed I think, but does not affect the ws_connect setting above
    script =  Script("""
                document.body.addEventListener('htmx:configRequest', (event) => {
                    event.detail.path = `""" +jc.asgi.ROOT_PATH +"""${event.detail.path}`
                })
                """)        
    return Titled('Websocket Test', cts, script)

async def on_connect(send):
    print("on connect")
    await send(Div('Hello, you have connected', id=nid))

async def on_disconnect( ):
    print('Disconnected!')

@app.ws('/ws', conn=on_connect, disconn=on_disconnect)
async def ws(msg:str, send):
    print("ws", msg)
    await send(Div('Hello ' + msg, id=nid))
    await sleep(2)
    return Div('Goodbye ' + msg, id=nid), mk_inp()

serve()
Py.Cafe

fasthtml-websocket-interaction

WebSocket Interaction with FastHTML
