django-spark - Event sourcing and handling

Version |release|

This is not supposed to be real documentation; it's more a reminder for myself.

The idea is that there are event sources and event handlers. Event sources may create a stream of spark.api.Event instances, where each event must have a group and a key. Additional data may be added to the Event as well. Keys are globally unique -- events with the same key are still only processed exactly once. Groups are used to determine which handlers handle a certain event.

Event handlers are functions which are called once per spark.api.Event instance if the event's group matches the event handler's regex.

Some usage example code

Given a challenge, create events for the challenge (the specifics do not matter):

from datetime import date
from spark import api

def events_from_challenge(challenge):
    if not challenge.is_active:
        return

    context = {"challenge": challenge}

    yield api.event("challenge_created", challenge.pk, context)

    if (date.today() - challenge.start_date).days > 2:
        if challenge.donations.count() < 2:
            yield api.event("challenge_inactivity_2d", challenge.pk, context)

    if (challenge.end_date - date.today()).days <= 2:
        yield api.event("challenge_ends_2d", challenge.pk, context)

    if challenge.end_date < date.today():
        yield api.event("challenge_ended", challenge.pk, context)

Send mails related to challenges (uses django-authlib's render_to_mail):

from authlib.email import render_to_mail

def send_challenge_mail(event):
    challenge = event["context"]["challenge"]
    render_to_mail(
        # Different mail text per event group:
        "challenges/mails/%s" % event["group"],
        {
            "challenge": challenge,
        },
        to=[challenge.user.email],
    ).send(fail_silently=True)

Register the handlers:

from spark import api

class ChallengesConfig(AppConfig):
    def ready(self):
        api.register_group_handler(
            handler=send_challenge_mail,
            group=r'^challenge',
        )

        Challenge = self.get_model('Challenge')

        # All this does right now is register a post_save signal
        # handler which runs the challenge instance through
        # events_from_challenge and processes the events:
        api.register_model_event_source(
            sender=Challenge,
            source=events_from_challenge,
        )

Now, events are generated and handled directly in process. Alternatively, you might want to handle events outside the request-response cycle. This can be achieved by only registering the model event source e.g. in a management command, and then sending all model instances through all event sources, and directly processing those events, for example like this:

from spark import api

api.register_model_event_source(...)

# Copied from the process_spark_sources management command inside
# this repository
for model, sources in api.MODEL_SOURCES.items():
    for instance in model.objects.all():
        for source in sources:
            api.process_events(api.only_new_events(source(instance)))

• Documentation
• Github