CQRS + Event Sourcing in Django: Ein praktischer Architekturansatz

✍️ Abschnitt 1: Einführung – Motivation & Zielsetzung

Quellcode & Repository

Der vollständige Quellcode dieses Projekts ist frei verfügbar auf GitHub unter github.com/dunkeltech/django-cqrs-example

Das Repository enthält:

• das Django-Projekt mit CQRS + Event Sourcing

• devenv.nix für eine sofort startbare Entwicklungsumgebung

• API-Test-Dateien für Bruno

• eine ausführliche README.md

• und alles, was du brauchst, um das Projekt lokal zum Laufen zu bringen oder für eigene Ideen weiterzuentwickeln

CQRS und Event Sourcing in Django – Warum?

Die meisten Django-Projekte entstehen aus einem klassischen Ansatz:
Ein Model, ein View, ein Form oder Serializer – und schon kann man Objekte erstellen, lesen, updaten oder löschen. Für viele Anwendungsfälle ist das auch vollkommen ausreichend.

Doch sobald ein Projekt komplexer wird, entstehen typische Herausforderungen:

• Lesende und schreibende Anforderungen unterscheiden sich stark.
  Beispiel: „Zeige eine aggregierte Liste aller Bestellungen nach Kunde“ vs. „Erstelle eine neue Bestellung mit Validierung und Geschäftslogik“.

• Zustandsänderungen sind nicht mehr nachvollziehbar.
  Wer hat wann was geändert? Warum steht der Datensatz heute so da? Klassische Audit-Felder reichen irgendwann nicht mehr aus.

• Fehlende Skalierbarkeit im Datenmodell.
  Ein ORM-Modell muss immer „alles“ können – lesen, schreiben, filtern, validieren – und wächst dadurch oft über seine Verantwortung hinaus.

Lösung: CQRS + Event Sourcing

Um diese Herausforderungen zu meistern, habe ich mich entschieden, mein Projekt auf Basis von CQRS und Event Sourcing aufzubauen:

• CQRS (Command Query Responsibility Segregation):
  Ich trenne strikt zwischen Commands (schreibende Vorgänge wie "erstelle Bestellung") und Queries (lesende Zugriffe auf z. B. Bestellübersichten).

• Event Sourcing:
  Ich speichere nicht nur den aktuellen Zustand, sondern alle Ereignisse, die zu diesem Zustand geführt haben. So kann ich jederzeit nachvollziehen, warum ein Objekt heute aussieht, wie es aussieht.

Ziel des Projekts

Das Ziel meines Projekts war es, eine schlanke, aber robuste Architektur zu bauen, die:

• sauber strukturiert ist und gut testbar bleibt

• Geschäftslogik in Commands kapselt

• lesende Zugriffe durch eigene Read-Modelle beschleunigt

• alle Änderungen als DomainEvent speichert

• sich bei Bedarf vollständig aus Events neu aufbauen lässt (Rebuild/Replay)

• und dabei Django & DRF weiterhin effizient nutzt

Stack & Werkzeuge

• Django + Django REST Framework als solides Fundament

• PostgreSQL als EventStore und ReadModel-Datenbank

• devenv.nix für ein reproduzierbares lokales Dev-Setup

• Bruno für automatisierte API-Tests

• pre-commit-Hooks für einheitlichen Code-Style

✍️ Abschnitt 2: Architekturüberblick – CQRS + Event Sourcing in Django

In diesem Abschnitt zeige ich dir, wie ich CQRS und Event Sourcing in Django konkret umgesetzt habe: Welche Schichten es gibt, wie sie zusammenspielen, und warum das auch mit einem klassischen Django-Stack erstaunlich gut funktioniert.

Die Grundidee: Trennung von Lesen und Schreiben

Das System ist entlang der CQRS-Prinzipien in zwei Hauptpfade aufgeteilt:

Commands (Write Path)

• Commands ändern den Zustand der Domäne

• Aber: Sie schreiben keine Daten direkt in Django-Modelle

• Stattdessen erzeugen sie Events, die den gewünschten Zustand beschreiben

• Diese Events werden im zentralen DomainEvent-Modell gespeichert

{
  "aggregate_type": "Order",
  "aggregate_id": "1234",
  "event_type": "OrderCreated",
  "payload": {
    "customer_name": "Alice",
    "product": "Monitor",
    "quantity": 2
  },
  "created_at": "2025-07-13T09:00:00Z"
}

Die wichtigsten Bausteine

DomainEvent

Das zentrale EventStore-Modell, das alle Änderungen speichert:

class DomainEvent(models.Model):
    aggregate_type = models.CharField(...)
    aggregate_id = models.CharField(...)
    event_type = models.CharField(...)
    payload = models.JSONField()
    created_at = models.DateTimeField(auto_now_add=True)

Commands

Kleine Klassen, die spezifische Anwendungsfälle abbilden, z. B.:

• CreateOrderCommand

• UpdateOrderCommand

• DeleteOrderCommand

Jeder Command erzeugt genau ein DomainEvent.

Projektionen (ReadModel-Sync)

Sobald ein Event gespeichert wird, wird es per Signal an eine Projektion weitergeleitet, z. B.:

class OrderProjection:
    def apply(self, event):
        match event.event_type:
            case "OrderCreated":
                self._handle_created(event)
            case "OrderUpdated":
                self._handle_updated(event)
            case "OrderDeleted":
                self._handle_deleted(event)

Queries

Einfach strukturierte Klassen, die direkt auf das ReadModel zugreifen, z. B.:

class GetOrderReadQuery:
    def execute(self, order_id):
        return OrderReadModel.objects.get(order_id=order_id)

Vorteile dieser Architektur

Vorteil	Beschreibung
Zustand rekonstruierbar	Events können jederzeit erneut abgespielt werden
Testbarkeit	Commands und Projections können separat getestet werden
Auditierbarkeit	Jedes Event bleibt nachvollziehbar gespeichert
Performance	Queries laufen über optimierte Read-Modelle, nicht über Events
Erweiterbar	Neue Felder oder Aggregates brauchen keine Migrations – nur neue Events

Wichtig zu verstehen

• Das ReadModel ist eine Sicht auf den Zustand, keine Quelle der Wahrheit

• Nur Events gelten als tatsächliche, unveränderbare Wahrheit

• Es ist möglich (und gewollt), das gesamte ReadModel aus Events neu aufzubauen (rebuild_read_models)

✍️ Abschnitt 3: Technische Umsetzung in Django

In diesem Abschnitt zeige ich dir, wie ich die in Abschnitt 2 beschriebene Architektur konkret mit Django und Django REST Framework (DRF) umgesetzt habe – inklusive Commands, Events, Projections und der API-Schicht.

1. DomainEvent – der EventStore

Zentrales Modell für alle Änderungen, die im System passieren – unabhängig vom Aggregat (z. B. Order, Customer, Product):

class DomainEvent(models.Model):
    aggregate_type = models.CharField(max_length=100)
    aggregate_id = models.CharField(max_length=64)
    event_type = models.CharField(max_length=100)
    payload = models.JSONField()
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        indexes = [
            models.Index(fields=["aggregate_type", "aggregate_id", "created_at"]),
        ]

2. Commands – Use Cases kapseln

Commands sind Klassen, die Geschäftslogik ausführen und dabei ein oder mehrere Events erzeugen.

Beispiel: CreateOrderCommand:

class CreateOrderCommand:
    def execute(self, data):
        order_id = str(uuid.uuid4())
        DomainEvent.objects.create(
            aggregate_type="Order",
            aggregate_id=order_id,
            event_type="OrderCreated",
            payload=data,
        )
        return order_id

3. Projektion – das ReadModel aktuell halten

Wenn ein neues Event gespeichert wird, reagieren Signal-Handler und rufen die passende Projektion auf.

class OrderProjection:
    def apply(self, event):
        match event.event_type:
            case "OrderCreated":
                self._handle_created(event)
            case "OrderUpdated":
                self._handle_updated(event)
            case "OrderDeleted":
                self._handle_deleted(event)

4. Signals – Event getriggert → Projektion anwenden

Hier verwende ich die Django-Signale um über Änderungen informiert zu werden und das ReadModel aufzubauen.

@receiver(post_save, sender=DomainEvent)
def handle_domain_event(sender, instance, **kwargs):
    if instance.aggregate_type == "Order":
        OrderProjection().apply(instance)

5. Queries – Zugriff aufs ReadModel

Für alle Lesezugriffe nutze ich einfache Query-Klassen:

class GetOrderReadQuery:
    def execute(self, order_id):
        try:
            return OrderReadModel.objects.get(order_id=order_id)
        except OrderReadModel.DoesNotExist:
            return None

6. DRF API – mit Create, Update, Delete

Die OrderAPIView unterstützt:

• GET /orders/ – alle Orders

• GET /orders/<id>/ – einzelne Order

• POST /orders/ – Bestellung erstellen (→ CreateOrderCommand)

• PATCH /orders/<id>/ – Bestellung aktualisieren (→ UpdateOrderCommand)

• DELETE /orders/<id>/ – Bestellung löschen (→ DeleteOrderCommand)

DRF-Serializer sorgen für Validierung:

class OrderUpdateSerializer(serializers.Serializer):
    customer_name = serializers.CharField(max_length=100, required=False)
    product = serializers.CharField(max_length=100, required=False)
    quantity = serializers.IntegerField(min_value=1, required=False)

ReadModel-Rebuild

Das ReadModel kann jederzeit neu aus dem EventStore aufgebaut werden:

python manage.py rebuild_read_models

Ergebnis

Man bekommt:

• stabile, nachvollziehbare Command-Pfade

• schnelle, einfache Query-Zugriffe

• vollständige Historie durch Events

• klare Trennung von Anwendungslogik und Persistenz

✍️ Abschnitt 4: Developer Experience & Setup

Eine Architektur ist nur dann produktiv nutzbar, wenn auch das Entwicklungserlebnis (Developer Experience) gut ist. Deshalb habe ich das Projekt so gestaltet, dass es sich mit einem einzigen Befehl sauber starten lässt – inklusive:

• Python-Umgebung

• PostgreSQL-Datenbank

• Codequalitätstools

• API-Testumgebung

1. Reproduzierbares Setup mit devenv.nix

Ich verwende devenv.sh auf Basis von Nix, um eine konsistente Entwicklungsumgebung bereitzustellen – unabhängig vom Betriebssystem oder Setup des Entwicklers.

Beispiel: devenv.nix

{
  packages = [ pkgs.git pkgs.openssl pkgs.python311Packages.pip ];
  languages.python = {
    enable = true;
    version = "3.11";
    venv.enable = true;
  };

  services.postgres = {
    enable = true;
    initialDatabases = [
      { name = "cqrs"; user = "cqrs"; pass = "cqrs"; }
    ];
  };

  pre-commit.hooks = {
    black.enable = true;
    isort.enable = true;
    flake8.enable = true;
  };
}

So startest du dein Dev-Setup

devenv up

2. API-Tests mit Bruno

Die API kann vollständig mit Bruno getestet werden – einem Open-Source-API-Client, ähnlich wie Postman, aber lokal und dateibasiert.

Ich habe alle relevanten API-Aufrufe in der Struktur api/ abgelegt:

• GET /orders/

• POST /orders/

• PATCH /orders/<id>/

• DELETE /orders/<id>/

→ Du kannst diese Dateien einfach öffnen und direkt testen, ohne manuelle Eingaben.

3. Codequalität via Pre-Commit

Ich nutze pre-commit mit folgenden Hooks:

• black für auto-formatting

• isort für sortierte Imports

• flake8 für Linting

pre-commit install

✍️ Abschnitt 5: Fazit – Was ich aus dem Projekt gelernt habe

Ein Architekturansatz mit echten Vorteilen

Die Kombination aus CQRS und Event Sourcing hat sich in meinem Django-Projekt als starkes Fundament für eine langfristig wartbare Backend-Architektur erwiesen.
Ich konnte damit:

• Geschäftslogik klar von Datenzugriff trennen

• jede Zustandsänderung transparent nachvollziehbar machen

• API-Zugriffe performanter und gezielter gestalten

• Projektionen gezielt auf UI- und Anwendungsbedürfnisse zuschneiden

Was besonders gut funktioniert hat

• Die Verwendung eines generischen DomainEvent-Modells statt spezialisierter Event-Tabellen hat das System schlank gehalten

• Mit einer einheitlichen Projection-Klasse ließ sich das ReadModel sehr klar verwalten

• Rebuild-Funktionalität aus Events hat das Vertrauen in die Architektur gestärkt

• DRF hat mir viel Zeit gespart bei API-Aufbau und Validierung

Herausforderungen und Learnings

• Signale und Transaktionen sauber zu synchronisieren, braucht etwas Sorgfalt (z. B. bei rebuild_read_models)

• Read-Projektionen müssen bei Schemaänderungen explizit aktualisiert werden (kein „magisches“ ORM)

• Man muss diszipliniert bleiben: Events sind unveränderlich, daher ist „Fixen“ alter Zustände nicht einfach (und auch nicht gewollt!)

Nächste Ausbaustufen

Das Projekt ist offen für viele sinnvolle Erweiterungen:

• Snapshots zur Beschleunigung bei großen Event-Ketten

• Event-Versionierung, wenn sich Payload-Strukturen ändern

• Webhooks / EventBus, um auf Events in anderen Systemen zu reagieren

• Admin-Interface für Event-Historie & Replay

Für wen ist dieser Ansatz geeignet?

Diese Architektur lohnt sich besonders, wenn du:

• komplexe Geschäftsregeln abbildest

• Versionierung & Nachvollziehbarkeit brauchst

• skalierbare APIs mit unterschiedlicher Lese- und Schreiblast aufbaust

• oder einfach ein bisschen mehr Ordnung und Klarheit im Backend willst

Danke fürs Lesen!

Wenn du dieses Projekt nachbauen oder erweitern möchtest, findest du im Repository alles, was du brauchst – inklusive README, devenv.nix, API-Tests und Lizenz.

Fragen, Anregungen oder eigene Erfahrungen? Ich freue mich über Feedback!