This is part of the Poddla series of posts.

This post is about Datastar and Django, I won’t go into depth about what they are, but here’s a short intro: With Datastar you can build backend driven web applications that feel as interactive as any Single Page Application (SPA, eg. React, Angular etc). It’s similar to HTMX but more tuned towards real time applications while being both smaller and more feature rich. Django is the premiere Python web application framework used to build traditional Multi Page Applications (MPA), REST services etc.

When I started Poddla, one of the things I wanted to explore was how to build a Datastar application using Django. In particular, I wanted to look at how to make it easier to build a Django + Datastar application that updates the DOM over SSE and does “fat morphs” (as described in The Tao of Datastar). There are other approaches to using Datastar as well, but this is the most exciting one.

I’ve used Datastar with both Django and FastAPI before, but I wanted to take a look at this “from scratch” and maybe try some new things as well.

Async Django

When using long-lived SSE requests you need to run Django in async mode with ASGI. Not only in production, but also when developing. I used Granian for this purpose. Here’s a post about Granian

Forms example

The Django forms library is great. It provides many niceties for handling form input, including input validation with a bunch of built-in validation functions for common input types and ways to write your own. It also supports rendering the HTML for the forms. Traditionally, if you used Django forms you had to give up a lot of interactivity, since they only rendered on the server as part of the normal form GET/POST flow, and hence you didn’t get any live as-you-type validation, feedback or other dynamic behavior. On the other hand, if you used a SPA, you had to define your validations twice, on the server and in the frontend. Datastar changes that.

Here’s the main body part of the settings page in Poddla:

It’s a regular “fat morph” style Datastar page, it keeps an SSE request open to Django that emits new HTML every time either the settings values in the database change or when this particular page’s state change (such as when one of the forms is opened/closed, form values are changed etc.). When the user modifies form values or submits the form, the page (via Datastar) makes HTTP calls that updates the state in Django.

If another user had submitted new settings, they would have instantly shown up on the page. At least as long as the form wasn’t open. It wouldn’t change form values from under the user’s feet.

How?

The view that streams new HTML to the browser needs to know when things change. There are multiple ways to think about this, and multiple technologies that could be used to communicate changes.

What I wanted was:

  1. A publish/subscribe (pub/sub) layer over Django model changes

    In this case I wanted to set up a pub/sub layer where when the Django ORM saves a model instance, subscribers to that model or model instance would get notified about it. When any part of the system changes a model instance, any other part of the system should be able to get to know about the change if they’re interested in it.

  2. A pub/sub layer for holding state for a page

    When the user turns a knob, applies a filter on a search page or types in a form input in such a way that the page needs re-rendering, we need to be able to store that state in an observable way.

In Django projects it’s quite common that code runs in multiple processes. That can be because of multiple web app workers, or because of a task queue like Celery that runs long running or periodic jobs in a different process. That means our pub/sub bus needs to be capable of working across different processes.

I’ve used NATS for this purpose before, but wanted to try Valkey (a Redis fork) this time.

Django model pub/sub

The solution I arrived at looks like this when used. Note that this is the first time I used this code in a project myself, and I will iterate on the library as I go. I’m not done. I hope I get feedback and improvement suggestions.

These code snippets are simplified and untested. The originals can be found in the Poddla repository.

Set up model publishers

First we set up the publishers in models.py:

from django.db import models

# `valkey_changes` is my Python module, not associated with the Valkey project.
# I need a better name for it.
from valkey_changes.model_publisher import ModelPublisher

podcast_publisher = ModelPublisher("podcasts:updates")
episode_publisher = ModelPublisher("episodes:updates")


class PodcastFeed(models.Model):
    url = models.URLField()
    name = models.TextField()


class Episode(models.Model):
    podcast = models.ForeignKey(PodcastFeed, on_delete=models.CASCADE, related_name="episodes")
    title = models.TextField()
    url = models.URLField()


podcast_publisher.register(PodcastFeed)
# When an episode changes, we publish as if the parent podcast changed:
podcast_publisher.register(Episode, resolve_pk=lambda instance: instance.podcast_id)

episode_publisher.register(Episode)

The resolve_pk argument on the register method makes it possible to publish that the podcast has changed when an episode is created/updated.

Subscribe to model changes

Then in views or wherever you need to listen to changes to the models you can do things like this:

from valkey_changes.changes import changes
from .models import (
    PodcastFeed,
    episode_publisher,
    podcast_publisher,
)

# Listen to changes to any `PodcastFeed` or `Episode`:
async with changes(podcast_publisher.subscribe()) as changed:
    while True:
        await changed.wait()
        # Do the work.

# Or you can listen to changes to only the `PodcastFeed` with id 1 (or any of its `Episode`s):
async with changes(podcast_publisher.subscribe(pk=1)) as changed:
    pass

# `changes` accepts multiple subscriptions:
async with changes(podcast_publisher.subscribe(), episode_publisher.subscribe()) as changed:
    pass

ModelPublisher publishes only the fact that a thing changed. It does not publish the full model instance, subscribers will have to fetch model instances from the database themselves if they need them.

Django page state

Back to the forms.

To hold and pub/sub local state for a browser tab I came up with this API:

Again, these code snippets are simplified and untested. Working code can be found in the Poddla repository.

You define a msgspec.Struct class that holds the state you need for a given page. It gets a secret ID (tab_id), optionally scoped to the user’s ID, that you have to pass to the SSE view and any “action” view that modifies the state. The tab_id should never be shared across browser tabs. The purpose of the state is to hold state for a single browser tab. The state is stored in Valkey, more about that later. By default the state expires and is deleted from Valkey 24 hours after it was last updated, but that time is configurable. The state should only ever be updated from a single action view, and the page should avoid concurrent requests to it, either by using loading overlays that block user interaction or by request cancellation (new request cancels old ones). This is easily achieved with Datastar, and the repo has examples of it.

A class that holds state for a browser tab

import msgspec

class FeedUpdateState(msgspec.Struct):
    tab_id: str
    show_form: bool = False
    data: dict[str, str | int] | None = None
    loading: bool = False
    can_save: bool = False
    save: bool = False

    def to_signals(self):
        """Returns the state as a dict that can be used as a Datastar signal."""
        data = self.data or {}
        return {
            "tab_id": self.tab_id,
            "show_form": self.show_form,
            "data": {"hours": data.get("hours"), "minutes": data.get("minutes")},
            "loading": self.loading,
            "can_save": self.can_save,
            "save": self.save,
        }

A store that holds state for a browser tab, with pub/sub

from valkey_changes.state_store import StateStore
feed_update_store: StateStore[FeedUpdateState] = StateStore(
    FeedUpdateState,
    namespace="settings_feed_update",
    default_factory=lambda tab_id: FeedUpdateState(tab_id=tab_id),
)

Examples of how to use the state and the store:

Create a new state

This would be done by the main view that later initiates the SSE updates stream.

When a user_id argument is provided like below, the same user id will have to be provided when subscribing to and updating the same state, or those calls will fail.

feed_update_state = await feed_update_store.new(user_id=request.user.id)

Subscribe to a state

This would be done by the SSE view.

Note that we’re subscribing to both the tab state and a settings model in the same changes call.

from valkey_changes.changes import changes

tab_feed = feed_update_store.subscribe(feed_update_id, user_id=request.user.id)

async with changes(tab_feed, settings_publisher.subscribe()) as changed:
    await changed.wait()
    state = tab_feed.state
    # Do work.

Get a state and update it

state = await feed_update_store.get(tab_id, request.user.id)
state.show_form = True
await feed_update_store.save(tab_id, state, user_id=request.user.id)

Using the state with a Django form

It isn’t as streamlined as it perhaps could be made to be. I need to use this more to arrive at a good abstraction, assuming there is a good one to be found.

Snippets from Poddla’s settings view:

# A class method to pull out the form values from a model instance (`current_settings`)
state.data = FeedUpdateForm.initial_from_settings(current_settings)
# But you could do things like this just as well:
state.data = {"hours": ..., "minutes": ...}

# Validate
form = FeedUpdateForm(data=state.data)
form.is_valid()


# The state as signals to the frontend:
state.to_signals()

The state needs to be sent to the frontend. The to_signals() method we added to FeedUpdateState earlier can be used for that.

From an SSE view using the Datastar Python SDK:

yield ServerSentEventGenerator.patch_signals(signals=state.to_signals())

Or from the HTML:

<!-- in the view: state_json = json.dumps(state.to_signals()) -->
<div data-signals__ifmissing="{{ state_json }}"></div>

And you have to bind (data-bind) the signals to their input elements manually:

<input type="number" min="0" data-bind="data.hours" aria-label="Hours" />

TODO

  1. The input elements have to be hooked up to signals manually. It might be worth it to create a “DatastarSignalsForm” subclass of forms.Form that does the signals binding automatically.
  2. The state and the forms are quite disconnected now, maybe something could be done there. I did briefly try some things with that, but wasn’t happy with it.
  3. As I use this code more I’ll break it out into its own library.

Full working examples of this can be found in the Poddla repository.

Details

Valkey-GLIDE is used as the Valkey client library. I’ve used it with Valkey, but the documentation says it’s compatible with Redis as well. My valkey_changes library strives to use a single Valkey connection, but there is some hairiness regarding that when it comes to different event loops.