Example single-file Python Nanodjango app using HTMX websockets over ASGI

nanodjango_htmx_websocket_example.py

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "nanodjango",
#     "channels",
#     "daphne",
#     "htpy"
# ]
# ///

from nanodjango import Django
import json
from channels.generic.websocket import WebsocketConsumer
from django.http import HttpResponse
from django.urls import path
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack

from htpy import (
    body,
    button,
    div,
    form,
    h1,
    head,
    html,
    input,
    meta,
    p,
    script,
    link,
    title,
    main,
    style,
    fieldset,
    article,
)


def html_template():
    return html[
        head[
            meta(charset="utf-8"),
            meta(name="viewport", content="width=device-width, initial-scale=1"),
            title["Nanodjango HTMX Websocket Example"],
            script(src="https://unpkg.com/[email protected]"),
            script(src="https://unpkg.com/htmx.org/dist/ext/ws.js"),
            link(
                rel="stylesheet",
                href="https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css",
            ),
            style[
                """
                .message { padding: .5rem;  }
                .message.user-message {
                    border: 1px solid #999;
                    border-radius: 0.5rem;
                    margin-bottom: 0.5rem;
                }
                .message.echo-message { 
                    font-weight: bold; 
                    border: 1px solid green;
                    border-radius: 0.5rem;
                    margin-bottom: 1rem;
                }
                """,
            ],
        ],
        body[
            main(class_="container")[
                article[
                    h1["Nanodjango + HTMX Websocket + htpy Example"],
                    p["Type a message and it will be echoed back via websocket."],
                    div(hx_ext="ws", ws_connect="/ws/echo/")[
                        div("#message-list"),
                        form(ws_send=True)[
                            fieldset(role="group")[
                                input(
                                    name="message",
                                    type="text",
                                    placeholder="Type your message...",
                                    autocomplete="off",
                                ),
                                button(
                                    class_="primary outline",
                                    type="submit",
                                    onclick="setTimeout(() => this.closest('form').querySelector('input[name=message]').value = '', 0)",
                                )["↩"],
                            ]
                        ],
                    ],
                ],
            ],
        ],
    ]


def response_message(message_text, is_user=False):
    return div("#message-list", hx_swap_oob="beforeend")[
        div(class_=["message", {"echo-message": not is_user, "user-message": is_user}])[
            message_text
        ]
    ]


app = Django(
    # EXTRA_APPS=[
    #     "channels",
    # ],
    #
    # Nanodjango doesn't yet support prepending "priority" apps to INSTALLED_APPS,
    # and `daphne` must be the first app in INSTALLED_APPS.
    INSTALLED_APPS=[
        "daphne",
        "django.contrib.admin",
        "django.contrib.auth",
        "django.contrib.contenttypes",
        "django.contrib.sessions",
        "django.contrib.messages",
        "whitenoise.runserver_nostatic",
        "django.contrib.staticfiles",
        "channels",
    ],
    CHANNEL_LAYERS={
        "default": {
            "BACKEND": "channels.layers.InMemoryChannelLayer",
        },
    },
    ASGI_APPLICATION="__main__.htmx_websocket_app",
)


@app.route("/")
def index(request):
    return HttpResponse(html_template())


class EchoConsumer(WebsocketConsumer):
    def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message_text = text_data_json.get("message", "")
        if not message_text.strip():
            return

        user_message_html = response_message(f"👤 {message_text}", is_user=True)
        self.send(text_data=user_message_html)

        echo_message_html = response_message(f"💻 You said: “{message_text}”")
        self.send(text_data=echo_message_html)


websocket_urlpatterns = [
    path("ws/echo/", EchoConsumer.as_asgi()),
]

# Configure ASGI application. Channels
htmx_websocket_app = ProtocolTypeRouter(
    {
        "http": app.asgi,
        "websocket": AuthMiddlewareStack(URLRouter(websocket_urlpatterns)),
    }
)


if __name__ == "__main__":
    app.run()