Live Components with Django and htmx

Using Django and htmx to render components through server-sent events.

Live Components with Django and htmx

I discovered django-components late last year and I quickly realized it was the missing piece in my Django+htmx workflow. It made my developer experience so much better, that I even started contributing to it.

django-components lets you build components that combine HTML, JS, and CSS in a single place. Plus, it now lets you use components as views. This feature allows you to keep all the logic for a part of your application in one place, giving you great locality of behavior (LOB).

A click-to-load component would look something like this:

from django.core.paginator import Paginator
from django_components import component

from src.app.models import Contact


@component.register("click_to_load")
class ClickToLoadTableComponent(component.Component):
    template = """
        {% for contact in page_obj %}
            <tr> 
                <td>{{ contact.id }}</td>
                <td>{{ contact.first_name }} {{ contact.last_name }}</td>
                <td>{{ contact.email }}</td>
                <td>{{ contact.status }}</td>
            </tr>
            {% if forloop.last and page_obj.has_next %} 
                <tr id="replaceMe">
                    <td colspan="4">
                        <button 
                            class='primary' 
                            hx-get="{% url 'contacts' page=page_obj.next_page_number %}"
                            hx-target="#replaceMe"
                            hx-swap="outerHTML">
                            Load more...
                        </button>
                    </td>
                </tr>
            {% endif %}
        {% endfor %}
    """

    def get_context_data(self, page_obj, **kwargs):
        return {"page_obj": page_obj}

    def get(self, request, page, **kwargs):
        paginator = Paginator(Contact.objects.order_by("id"), 3)
        page_obj = paginator.get_page(page)
        context = {"page_obj": page_obj}
        return self.render_to_response(context)

You can use this component in any view using {% component 'click_to_load' page_obj=page_obj %} or render it outside of a view by adding it to urls.py:

from django.urls import path

from src.components.click_to_load.table import ClickToLoadTableComponent

urlpatterns = [
    path(
        "contacts/<int:page>",
        ClickToLoadTableComponent.as_view(),
        name="contacts",
    ),
]

Short and sweet, just like the best things in life.

Django Live Components

But I thought it would be fun to use the library for something it wasn't designed for: streaming component changes through server-sent events (SSE).

It took me a few hours and several reads of Víðir's tutorial to figure it out, but it worked. It's just a bit hacky. All the pieces were there. I just had to find a way to put them together.

The code is available here.

I had a simple idea: set up a Redis pub/sub channel for server notifications. When the client loads the page, it subscribes to this notification channel. Each time the server publishes a new notification, the system reads it from the channel. Then, it renders the HTML and sends it to the client using Server-Sent Events (SSE).

First, you need a notification component, with a streaming view that updates the client whenever a new notification occurs, and a way to subscribe to new notifications sent from the server.

Here's what I came up with:

# src/components/notification.py

import asyncio
import json
from typing import AsyncGenerator
import redis.asyncio as redis
from django.http import StreamingHttpResponse
from django_components import component

r = redis.from_url("redis://localhost")


def sse_message(event_id: int, event: str, data: str) -> str:
    data = data.replace("\n", "") # remove new lines or it won't work
    return f"id: {event_id}\n" f"event: {event}\n" f"data: {data.strip()}\n\n"


class NotificationComponent(component.Component):
    template = """
    <div style="color: {{color}};" role="alert">
        <span style="font-weight: bold;">{{ title }}</span> {{ message }} 
    </div>
    """


notification_component = NotificationComponent(
    registered_name="notification",
)


async def streaming_response(*args, **kwargs) -> AsyncGenerator[str, None]:
    async with r.pubsub() as pubsub:
        await pubsub.subscribe("notifications_channel")
        try:
            while True:
                message = await pubsub.get_message(
                    ignore_subscribe_messages=True, timeout=1
                )
                if message is not None:
                    notification_data = json.loads(message["data"].decode())
                    sse_message_rendered = sse_message(
                        notification_data["id"],
                        "notification",
                        notification_component.render(
                            {
                                "title": notification_data["title"],
                                "message": notification_data["message"],
                                "color": notification_data["color"],
                            }
                        ),
                    )
                    yield sse_message_rendered
                await asyncio.sleep(0.1) # give control back to the event loop, otherwise tasks are killed
        finally:
            await r.aclose()


async def streaming_view(request):
    return StreamingHttpResponse(
        streaming_content=streaming_response(),
        content_type="text/event-stream",
    )

And you should include this in your urls.py:

from django.urls import path
from components.notification import streaming_view

urlpatterns = [
    path(
        "notification/",
        streaming_view,
        name="stream_notification",
    ),
]

Then, you need a simple HTML template to show these notifications. I used the htmx SSE extension to handle the SSE connection on the client. This was my template:

<!-- src/templates/index.html -->
<!DOCTYPE html>
<html lang="en">
    <head>
        <meta charset="UTF-8">
        <meta name="viewport" content="width=device-width, initial-scale=1.0">
        <title>Django Live Components</title>
    </head>
    <body>
        <div hx-ext="sse"
             sse-connect="{% url 'stream_notification' %}"
             sse-swap="notification"></div>
        <script src="https://unpkg.com/htmx.org@1.9.10" integrity="sha384-D1Kt99CQMDuVetoL1lrYwg5t+9QdHe7NLX/SoJYkXDFfX37iInKRy5xLSi8nO7UC" crossorigin="anonymous">
        </script>
        <script src="https://unpkg.com/htmx.org/dist/ext/sse.js"></script>
    </body>
</html>

Finally, you need a script to simulate these server notifications:

# random_notifications.py
import redis
import json
import random
import time

REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_CHANNEL = "notifications_channel"

r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)


def create_random_notification():
    """Create a random notification message"""
    return {
        "id": random.randint(1, 1000),
        "title": "Notification " + str(random.randint(1, 100)),
        "message": "This is a random message " + str(random.randint(1, 100)),
        "color": random.choice(["blue", "green", "red", "black", "gray", "purple"]),
        "timestamp": time.ctime(),
    }


def publish_notification():
    """Publish a random notification to the Redis channel"""
    notification = create_random_notification()
    r.publish(REDIS_CHANNEL, json.dumps(notification))
    print(f"Published: {notification}")


if __name__ == "__main__":
    try:
        while True:
            publish_notification()
            time.sleep(3)
    except KeyboardInterrupt:
        print("Stopped notification publisher")

You can run Redis on Docker to run this script. It'll start adding notifications to the Redis channel.

When you run your server in development mode, you can see the live component in action. It should look like this:

0:00
/0:05

This was fun to try, but I have to say, Dr. Ian Malcolm's quote from Jurassic Park crossed my mind a few times: "They were so preoccupied with whether they could, they didn't stop to think if they should."