Server sent events#

While we don’t have any direct support for websockets or SSE components out the box. It is possible to build your own SSE component leveraging a few tools.

You can find a couple of examples of this in our demos (TODO add links).

Using SSE in your project#

To use SSE you need django_eventstream and an event service such as pushpin.

Install django_eventstream#

pip install django_eventstream

Add the following to your settings file:

INSTALLED_APPS = [
    ...
    'django_eventstream',
]

MIDDLEWARE = [
    'django_grip.GripMiddleware',
     ...
]

GRIP_URL = 'http://localhost:5561'
EVENTSTREAM_ALLOW_ORIGIN = "http://127.0.0.1:8000"

Add url pattern for SSE channels

dashboard_channels = []  # per dashboard, per component etc
urlpatterns += [path('events/', include(django_eventstream.urls), {"channels": dashboard_channels})]

Additional notes:

Refer to django_eventstream regarding authentication
This is posting all events to test, you’d want to consider spreading events to different channels for dashboards/components e.g. https://github.com/fanout/django-eventstream#routes-and-channel-selection

Run pushpin#

For our demo we use pushpin to aid with event distribution. The easiest way to start pushpin locally is to add it to a docker compose i.e.

pushpin:
  image: fanout/pushpin
  environment:
    target: '127.0.0.1:8000'
  volumes:
    - "./pushpin.conf:/etc/pushpin/:rw"
  network_mode: host

where pushpin.conf/ is a folder containing a routes file with the following:

* 127.0.0.1:8000,over_http

Additional notes:

For production - you will also need to deploy a pushpin instance or use Fanout Cloud (see below).

Add your SSE components#

Once eventstream and pushpin are setup you can create a component that accepts SSE. In our demo we use the following to make a couple of existing components SSE ready. Largely this is a new template which will leverage HTMX and a property to get the pushpin URL into the template.

from dashboards.component import Chart, Stat
from dataclasses import dataclass
from typing import Optional

@dataclass
class SSEStat(Stat):
    template: str = "dashboards/components/sse_stat.html"
    poll_rate: Optional[int] = 10

    @staticmethod
    def pushpin_url():
        """
        Assuming docker pushpin is running, in real world this would be proxied to application.
        """
        return "http://localhost:7999/events/"


@dataclass
class SSEChart(Chart):
    template: str = "dashboards/components/sse_chart.html"
    poll_rate: Optional[int] = None

    @staticmethod
    def pushpin_url():
        """
        Assuming docker pushpin is running, in real world this would be proxied to application.
        """
        return "http://localhost:7999/events/"


...

class DemoDashboard(Dashboard):
    sse_stat = SSEStat()
    ...

At a template level, taking SSEStat as an example, we can the built in SSE features in HTMX to connect to pushpin

<div hx-ext="sse" sse-connect="{{ component.pushpin_url }}" sse-swap="{{ component.template_id }}">
  Contents of this box will be updated in real time
  with every SSE received.
</div>

In order for events to be sent you could either have a cron job, management command, celery task or one even one of our django-pipelines. For example in our demo, we render the Stat fully into an event:

value = randint(1, 100)
sse_stat = Stat(
    value={"text": f"{value}%", "sub_text": "Via SSE"}, key="sse_stat"
)
rendered_stat = render_to_string(
    sse_stat.template,
    {
        "component": sse_stat,
        "rendered_value": sse_stat.get_value(),
    },
)

parsed_stat = strip_spaces_between_tags(rendered_stat.strip())

# Here we return a actual rendered stat component, but this could be simply a value JSON also
send_event("test", "sse_stat", parsed_stat, json_encode=False)

This will send a stat via an event to the sse_stat component.