Using Mercure with Django for Server-sent events

mercure.py

# Helper functions for publishing events, generating JWT tokens, and generating the Hub URL.

def publish_event(event_type, topic, targets):
    """
    Publishes an event to the Mercure Hub.
    event_type: The type of the event, can be any string
    topic: The topic the event will be sent to. Only subscribers who request this topic will get notified.
    targets: The targets that are eligible to get the event.
    """
    token = get_jwt_token([], targets)
    headers = {
        'Authorization': 'Bearer {}'.format(token),
        'Content-Type': 'application/x-www-form-urlencoded',
    }
    data = {
        'type': event_type,
        'topic': topic,
        # Mercure expects this data field, even though we don't need it
        # for Intercooler updates.
        'data': '{}',
    }
    requests.post(
        settings.MERCURE_HUB_URL,
        data=data,
        headers=headers,
    )


def get_jwt_token(subscribe_targets, publish_targets):
    """
    Creates a Mercure JWT token with the subscribe and publish targets.
    The JWT token gets signed with a key shared with the Mercure Hub.
    """
    return jwt.encode(
        {
            'mercure': {
                'subscribe': subscribe_targets,
                'publish': publish_targets
            }
        },
        settings.MERCURE_JWT_KEY,
        algorithm='HS256'
    )


def get_hub_url(topics):
    """
    Returns the URL used to subscribe to the given topics in Mercure. The response
    will be an event stream.
    """
    params = [('topic', t) for t in topics]
    return settings.MERCURE_HUB_URL + '?' + urllib.parse.urlencode(params)

mixins.py

# View mixin used for views that will subscribe to real-time events

class MercureMixin(object):
    """
    This view mixin will add a Set-Cookie header to the response. This cookie will
    include authorization information for the Mercure Hub in the form of a JWT token.
    The view needs to implement the subscribe and publish targets.
    """

    # Views that need to subscribe to events on the client should override this
    # attribute with the targets to subcribe to.
    mercure_subscribe_targets = []

    # Views that need to publish events from the client should override this
    # attribute with the targets to publish to.
    mercure_publish_targets = []

    # Views that need to subscribe to events should override this attribute
    # with the topics to subscribe to.
    mercure_hub_topics = []

    def get_mercure_subscribe_targets(self):
        """
        If the view needs to dynamically determine subscribe targets, it can
        override this method.
        """
        return self.mercure_subscribe_targets

    def get_mercure_publish_targets(self):
        """
        If the view needs to dynamically determine publish targets, it can
        override this method.
        """
        return self.mercure_publish_targets

    def get_mercure_hub_topics(self):
        """
        If the view needs to dynamically determine topics, it can
        override this method.
        """
        return self.mercure_hub_topics

    def dispatch(self, request, *args, **kwargs):
        """
        If Mercure is enabled, we will set a cookie in the response with
        a JWT token used for authentication/authorization with the Mercure Hub.
        Connections to the hub from the client will automatically pass along this
        cookie.
        """
        response = super(MercureMixin, self).dispatch(request, *args, **kwargs)

        if settings.MERCURE_ENABLED:
            token = mercure.get_jwt_token(
                self.get_mercure_subscribe_targets(),
                self.get_mercure_publish_targets()
            )
            response.set_cookie(
                'mercureAuthorization',
                token,
                httponly=True,
                domain=settings.MERCURE_HUB_COOKIE_DOMAIN,
                path='/',
                secure=settings.MERCURE_HUB_SECURE_COOKIE,
            )

        return response

    def get_context_data(self, **kwargs):
        """
        Adds the Mercure Hub URL to the template context. Views can use this URL to
        connect to the hub from the client. For Intercooler support, this URL should be
        set on a container HTML element using the `ic-sse-src` attribute.
        """
        context = super(MercureMixin, self).get_context_data(**kwargs)
        if settings.MERCURE_ENABLED:
            context['mercure_hub_url'] = mercure.get_hub_url(self.get_mercure_hub_topics())
        return context

status_detail.html

{# partial that renders a single user status and updates itself when getting a Server-sent event #}
<div ic-trigger-on="sse:status-updated-{{ user.id }}" ic-src="{% url 'status-detail' pk=user.pk %}">
  User: {{ user.name }} Status: {{ user.status }}
</div>

status_list.html

{# page that shows a list of users. The user status will get real-time updates #}
{# The div below subscribes to the Mercure hub #}
<div ic-sse-src="{% mercure_hub_url %}">
  {% for user in user_list %}
    {% include 'status_detail.html' %}
  {% endfor %}
</div>

views.py

# Example view that will get real-time events on the client

class StatusList(MercureMixin, generic.ListView):
  template_name = "status_list.html"
  queryset = User.objects.all()
  mercure_subscribe_targets = ["admin"]
  mercure_hub_topics = ["status"]
  
class StatusDetail(MercureMixin, generic.DetailView):
  template_name = "status_detail.html"
  model = User