In einem vorherigen Beitrag habe ich bereits über das Thema Event Sourcing geschrieben. Heute möchte ich eine konkrete mögliche Implementierung zeigen.

Der Fokus dieser Implementierung liegt dabei auf starke Typisierung sowie hohe Code-Qualität. Hierbei verwende ich sowohl Linter, als auch Unit Tests.

Der vollständige Quellcode der hier beschriebenen Bibliothek ist über das Code-Repository für diesen Blogpost herunterladbar.

Anforderungen

Es gibt eine schöne Lehre aus der Software-Entwicklung:

"Weeks of coding can save you hours of planning." - Unknown

Daher möchte ich erstmal grob planen, was ich eigentlich bauen will, beziehungsweise was das Endergebnis sein soll:

• Eine wiederverwendbare Bibliothek, welche häufig benötige Funktionen für Anwendungen enthalten soll, die Event Sourcing verwenden.

• Die Bibliothek soll agnostisch gegenüber der Speicherung der Events sein, und darf daher gewisse Annahmen nicht treffen (etwa wie die ID-Generierung oder Event-Serialisierung funktioniert).

• Der Code muss modernen Entwicklungs-Standards gerecht werden, dies wird durch mehrere Werkzeuge sichergestellt.

Qualitäts-Sicherung

Ich verwende drei Werkzeuge bzw. Bibliotheken, um die Qualität des Codes dauerhaft sicher zu stellen:

• Mypy: Bereit seit mehreren Versionen von Python3 gibt es die Möglichkeit, optionale Type Hints bzw. Annotations Code hinzuzügen, um anzuzeigen welche Datentypen z. B. eine Funktion erwartet und wieder zurück gibt. Diese werden jedoch zur Laufzeit nicht erzwungen, können dafür jedoch im Vorfeld durch Mypy geprüft werden.

• Pylint: Überprüft Python-Code statisch auf Fehler und Probleme, wie etwa undefinierte Variablen oder fehlende Dokumentations-Kommentare.

• pyTest: Python hat zwar ein Unittest-Framework in der Standard-Bibliothek, dieses leidet jedoch unter einer recht unbequemen API und bietet weniger Möglichkeiten als pyTest mit seinen 315+ externen Plugins.

Diese Werkzeuge können zu verschiedenen Zeiten während des Entwicklungs-Prozesses ausgeführt werden:

• Manuell während der lokalen Entwicklung.
• Automatisch als git pre-commit hook, um sicherzustellen dass kaputter Code gar nicht erst ins Repository gelangt.
• Während des CI um sicherzustellen, dass kaputter Code niemals in die Produktions-Umgebung gelangt.

Vorbereitung

Im ersten Schritt lege ich, wie bei jedem Python-Projekt, ein virtualenv im Projektverzeichnis an. Im Anschluss können die notwendigen Pakete installiert werden.

python3 -m venv ./venv
./venv/bin/pip install mypy pylint pytest

API definieren

Als ersten Schritt in der Implementierung habe ich mir überlegt: Wie soll die API am Ende aussehen? Eine API ist praktisch ein Vertrag welcher sich später nur schwer ändern lässt, daher sollte in diesen Schritt durchaus etwas Zeit investiert werden.

Beim Herumspielen und Testen mit verschiedenen Möglichkeiten bin ich am Ende zu einer Schnittstelle gekommen, welche auf folgenden Gedanken basiert:

• Die Definition der Events (= Daten) soll von der Verarbeitung (= Business-Logik) getrennt sein. Dies erlaubt verschiedenen Arten von Event-Consumern, die Ereignisse auf unterschiedliche Weise verarbeiten können - aber dennoch die gleichen Event-Typen verwenden können.

• Die Klasse welchen den Zustand und das Event-Handling verwaltet, soll sich nur um reine Events kümmern. Der ganze weitere Verwaltungs-Aufwand, etwa IDs oder Zusatzinformationen wie "Welcher Benutzer war das?" gehören in eine weitere Schicht, da nicht jeder Anwendungsfall diese braucht.

• Im Idealfall erzwingt die Implementierung auch keine eigenen Basis-Klassen für Daten und Handler, was die Integration in bestehenden Code vereinfacht.

Die API

Final habe ich mich für die folgende API entschieden, welche ich beispielhaft erklären möchte. Hintergrund zur "Business-Logik" hier: Alle Beispiele gehen von einem einfachen Multiplayer-Spiel aus, in dem alle Spieler nacheinander agieren, etwa ein Quiz- oder Kartenspiel.

Daten

Die ersten beiden Events sind die Registrierung bzw. das Verlassen von Spielern. Dies sind einfache Datenklassen ohne Logik. Da der StateManager selbst keinerlei Annahmen zu den Event-Objekten macht (und diese einfach nur durchreicht) könnten dies auch komplexe Klassen sein - wobei ich jedoch davon abrate, hier komplexere Logik zu implementieren.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import uuid
from dataclasses import dataclass, field


@dataclass
class RegisterPlayer:
    """Registration of a new player in the game"""

    name: str
    player_id: str = field(default_factory=lambda: str(uuid.uuid4()))


@dataclass
class UnregisterPlayer:
    """Player is leaving the game"""

    player_id: str

Initialisierung

Im ersten Schritt wird ein Instanz des StateManager initialisiert, welche als ersten Parameter einen Funktion erhält, die einen "initialen Zustand" zurück gibt. Dies muss eine Funktion sein, da diese während des Lebenszyklus des StateManager mehrfach aufgerufen werden kann (etwa beim Reset bzw. Undo).

Die State-Funktion muss übrigens kein dict zurück geben, komplexere Datenstrukturen sind auch möglich - für den StateManager ist der Zustand praktisch eine "schwarze Box", welcher nur durchgereicht wird.

1
2
3
4
5
6
7
8

def get_default_state():
    """Return the default state of a new empty game"""
    return {
        "players": {},
    }


manager = StateManager(get_default_state)

Handler

Im Anschluss werden die Handler für die einzelnen Events registriert. Wie bereits weiter oben beschrieben erlaubt die Trennung von Daten und Logik, dass verschiedene StateManager unterschiedliche Handler für Events haben.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# Als Decorator
@manager.handle(RegisterPlayer)
def on_register(state, event):
    state["players"][event.player_id] = {
        "name": event.name,
    }


# Falls Funktionen und StateManager getrennt werden
def on_unregister(state, event):
    state["players"].pop(event.player_id)
    manager.handle(UnregisterPlayer)(on_unregister)

Ein weiterer Vorteil dieser Art der Definition ist auch, dass die einzelnen Handler-Funktion einfach getestet werden können.

🤔  Man kann sich jetzt darüber streiten, ob es besser wäre immer einen neuen Zustand zurück zu geben (wie es etwa bei Redux der Fall ist), oder den Zustand direkt manipuliert wie hier. Ich bevorzuge letzteres, da dies schlicht performanter und simpler ist. Ein Hauptgrund bei Redux war die Möglichkeit des einfachen Zustand-Vergleichs anhand der Objekt-Identität, diese Funktion brauchen wir aber nicht.

Simpler Test

Im Anschluss erstellen wir ein paar Events und wenden diese auf den Zustand an:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

from pprint import pprint

from models import RegisterPlayer, UnregisterPlayer

# Zwei Spieler anlegen
manager.apply(
    RegisterPlayer(player_id="test", name="Mitch"),
    RegisterPlayer(name="Guest 1"),
)
pprint(manager.get())

# Einen wieder entfernen
manager.apply(UnregisterPlayer(player_id="test"))
pprint(manager.get())

# Entfernen rückgängig machen (löscht das letzte Event und berechnet den Zustand neu)
manager.undo()
pprint(manager.get())

Konkrete Implementierung

Im Code-Repository für diesen Blogpost kann die komplette Implementierung eingesehen werden. Im Gegensatz zu den Beispielen hat diese noch folgende zusätzliche Funktionen:

• Interne Speicherung der Event-Historie mit Möglichkeit, rückwärts zu laufen (genau genommen ein Replay aller Events bis zum Vorletzten).

• Fehlerbehandlung für den Fall, dass für ein Event kein Handler definiert ist (was je nach Anwendungsfall in Ordnung sein kann).

Unit Tests

Der kleine Beispiel-Test oben ist ja ganz nett, aber er deckt nicht die volle Funktionalität ab und die Ausführung ist manuelle Arbeit. Programmieren ist automatisieren, also schreiben wir einen ordentlichen Test!

Der Test basiert großteils auf den vorherigen Beispielen, jedoch mit einer Reihe von zusätzlichen assert-Statements um zu prüfen, ob die Klasse auch das tut was sie soll.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from events import RegisterPlayer, UnregisterPlayer
from state import StateManager, get_default_state


def test_register_unregister():
    manager = StateManager(lambda: {"players": {}})

    # Sicherstellen, dass Spieler sauber registriert werden
    manager.apply(RegisterPlayer(player_id="test", name="Mitch"))
    assert "test" in manager.get()["players"]

    # Sicherstellen, dass Spieler sauber entfernt werden
    manager.apply(UnregisterPlayer(player_id="test"))
    assert "test" not in manager.get()["players"]

Zusammengefasst

Um nicht ständig die gleichen Befehle wiederholen zu müssen, packe ich diese gerne in ein Makefile. Ich verwende gerne Make, da man meistens davon ausgehen kann, dass es auf Entwickler-Systemen schon installiert ist.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

lint:
    @echo "=============== Running mypy (Type Checks) ==============="
    ./venv/bin/mypy state_manager.py
    @echo

    @echo "=============== Running pylint (Code Checks) ==============="
    ./venv/bin/pylint state_manager.py
    @echo


test:
    @echo "=============== Running pytest (Unit Tests) ==============="
    ./venv/bin/pytest

Ergebnis der Ausführung beider Befehle:

🥳

But there is more!

Bei diesem kleinen Beispiel ist es noch recht einfach zu prüfen, ob wir unseren gesamten Code testen. Aber was ist mit einer Anwendung, die über 50 Dateien verteilt ist? Code coverage checks to the rescue!

./venv/bin/pip install pytest-cov

./venv/bin/pytest --cov-report term-missing --cov=state_manager

Die bemängelte Zeile ist der Code für den Fall, dass ein Event verarbeitet werden soll, für das keinen Handler definiert wurde. Dies kann mit einem weiteren Test abgefangen werden:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import pytest

from models import UnregisterPlayer
from state import StateManager, MissingHandlerError


def test_statemanager_missing_handler():
    """Make sure StateManager raises an error if handlers are missing."""
    manager = StateManager(lambda: None)
    with pytest.raises(MissingHandlerError):
        manager.apply(UnregisterPlayer(player_id="404"))

Integration in den git Workflow

Wie bereits am Anfang beschrieben kann es praktisch sein, die Ausführung der Lints und Tests vor einem git commit zu erzwingen. Dies stellt sicher, dass der Code im Repository immer "sauber" ist, und nicht zu Problemen bei anderen Entwicklerinnen oder im CI führt.

Dies kann über einen kleinen git-Hook eingerichtet werden.

#!/bin/sh
make lint || exit 1
make test || exit 1

Installation:

cp ./git-pre-commit-hook.sh ./.git/hooks/pre-commit

⚠️  Dies wirkt sich auf das ganze Repository aus. Falls in einem größeren (Mono-)Repository mehrere Projekte sind, müssen die Hooks entsprechend angepasst werden.

Abschluss

Die in diesem gezeigte Implementierung ist nur die niedrigste Schicht in einem größeren System: Für eine vollständige Anwendung von Event Sourcing sind weitere Funktionen notwendig, etwa die Serialisierung und Speicherung der Events.