Implement Event Sourcing with CQRS in Python

Overview

An auditor asks why account 42 showed a different balance on March 15 than it does today. With a CRUD table that history is gone — the last UPDATE won. Event sourcing answers the question because every state change is stored as an immutable event in an append-only log; the current balance is a projection, not the record. CQRS pairs with it: commands append events on the write side, and read models are built by replaying events into tables shaped for each query.

This recipe builds a working implementation in Python with PostgreSQL: an event store with real concurrency control, an aggregate root, a command handler, projections with durable idempotency, and snapshots for long streams. For a MySQL or SQL Server version of the event store, see the relational event sourcing recipe.

When to Use This

• Systems that must answer “what was the state at time X” (financial transactions, healthcare records, compliance-heavy domains)
• Read-heavy systems that need several read models over the same events (list views, dashboards, search indexes)
• Complex domains where replaying events is the fastest way to debug or test a state transition
• Event-driven architectures where the log doubles as the source of truth for other consumers

Skip event sourcing when:

• The domain is plain CRUD with no audit or temporal-query requirements; the infrastructure cost buys you nothing
• Nobody on the team can own event schema evolution and projection maintenance; both are ongoing work, not a one-time setup
• Users need read-your-own-writes consistency everywhere; projections are eventually consistent by design

Prerequisites

• Python 3.11+
• PostgreSQL (or any ACID-compliant database)
• sqlalchemy and pydantic packages

Companion code: the event-sourcing-cqrs-pattern companion repo has these examples as runnable files plus a main.py end-to-end demo.

Solution

1. Event Definitions

import uuid
from datetime import datetime, timezone
from pydantic import BaseModel, Field

class DomainEvent(BaseModel):
    # default_factory runs per instance — a plain default would be
    # evaluated once at class definition and shared by every event.
    event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    event_type: str
    aggregate_id: str
    version: int
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    metadata: dict = Field(default_factory=dict)

class AccountCreated(DomainEvent):
    event_type: str = "AccountCreated"
    owner_name: str
    initial_balance: float

class MoneyDeposited(DomainEvent):
    event_type: str = "MoneyDeposited"
    amount: float
    description: str

class MoneyWithdrawn(DomainEvent):
    event_type: str = "MoneyWithdrawn"
    amount: float
    description: str

class AccountClosed(DomainEvent):
    event_type: str = "AccountClosed"
    reason: str

2. Event Store

import json
from sqlalchemy import (
    create_engine, Column, String, Integer, DateTime, Text,
    UniqueConstraint, text,
)
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker, declarative_base
from typing import List, Optional

Base = declarative_base()

class EventRecord(Base):
    __tablename__ = 'event_store'
    __table_args__ = (
        # The real concurrency guard: two writers can't commit the same
        # (aggregate_id, version) pair even if both pass a SELECT check.
        UniqueConstraint('aggregate_id', 'version', name='uq_aggregate_version'),
    )
    id = Column(String, primary_key=True)
    aggregate_id = Column(String, nullable=False, index=True)
    event_type = Column(String, nullable=False)
    version = Column(Integer, nullable=False)
    timestamp = Column(DateTime(timezone=True), nullable=False)
    data = Column(Text, nullable=False)
    # 'metadata' is a reserved attribute name in Declarative — map it explicitly
    event_metadata = Column('metadata', Text, nullable=False, default='{}')

class SnapshotRecord(Base):
    __tablename__ = 'snapshots'
    aggregate_id = Column(String, primary_key=True)
    version = Column(Integer, nullable=False)
    state = Column(Text, nullable=False)
    created_at = Column(DateTime(timezone=True), nullable=False)

class ProcessedEvent(Base):
    __tablename__ = 'processed_events'
    event_id = Column(String, primary_key=True)
    processed_at = Column(DateTime(timezone=True), nullable=False)

class ConcurrencyError(Exception):
    pass

class EventStore:
    def __init__(self, database_url: str):
        self.engine = create_engine(database_url)
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)

    def append(self, aggregate_id: str, events: List[DomainEvent], expected_version: int):
        session = self.Session()
        try:
            # Fast-path check; the UNIQUE constraint above is the actual guarantee
            current = session.execute(
                text("SELECT MAX(version) FROM event_store WHERE aggregate_id = :aid"),
                {'aid': aggregate_id}
            ).scalar()
            current = current if current is not None else -1

            if current != expected_version:
                raise ConcurrencyError(
                    f"Expected version {expected_version}, got {current}"
                )

            for event in events:
                session.execute(
                    text("""INSERT INTO event_store (id, aggregate_id, event_type, version, timestamp, data, metadata)
                            VALUES (:id, :aid, :etype, :ver, :ts, :data, :meta)"""),
                    {
                        'id': event.event_id,
                        'aid': aggregate_id,
                        'etype': event.event_type,
                        'ver': event.version,
                        'ts': event.timestamp,
                        'data': event.model_dump_json(),
                        'meta': json.dumps(event.metadata),
                    }
                )

            session.commit()
        except IntegrityError:
            session.rollback()
            # A concurrent writer committed the same version first
            raise ConcurrencyError(f"Version conflict on aggregate {aggregate_id}")
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

    def get_events(self, aggregate_id: str, from_version: int = 0) -> List[DomainEvent]:
        session = self.Session()
        try:
            records = session.execute(
                text("""SELECT data FROM event_store
                        WHERE aggregate_id = :aid AND version > :ver
                        ORDER BY version ASC"""),
                {'aid': aggregate_id, 'ver': from_version}
            ).fetchall()

            events = []
            for record in records:
                data = json.loads(record[0])
                event_class = EVENT_REGISTRY.get(data['event_type'], DomainEvent)
                events.append(event_class(**data))
            return events
        finally:
            session.close()

EVENT_REGISTRY = {
    'AccountCreated': AccountCreated,
    'MoneyDeposited': MoneyDeposited,
    'MoneyWithdrawn': MoneyWithdrawn,
    'AccountClosed': AccountClosed,
}

3. Aggregate Root

from abc import ABC, abstractmethod
from typing import List

class AggregateRoot(ABC):
    def __init__(self, aggregate_id: str):
        self.id = aggregate_id
        self.version = -1
        self._pending_events: List[DomainEvent] = []

    @abstractmethod
    def apply(self, event: DomainEvent):
        pass

    def raise_event(self, event: DomainEvent):
        event.version = self.version + 1
        self.apply(event)
        self.version = event.version
        self._pending_events.append(event)

    def get_pending_events(self) -> List[DomainEvent]:
        return self._pending_events

    def clear_pending_events(self):
        self._pending_events.clear()

    @classmethod
    def from_events(cls, aggregate_id: str, events: List[DomainEvent]):
        aggregate = cls(aggregate_id)
        for event in events:
            aggregate.apply(event)
            aggregate.version = event.version
        return aggregate

class BankAccount(AggregateRoot):
    def __init__(self, aggregate_id: str):
        super().__init__(aggregate_id)
        self.owner_name: str = ""
        self.balance: float = 0
        self.status: str = "nonexistent"

    def apply(self, event: DomainEvent):
        if isinstance(event, AccountCreated):
            self.owner_name = event.owner_name
            self.balance = event.initial_balance
            self.status = "active"
        elif isinstance(event, MoneyDeposited):
            self.balance += event.amount
        elif isinstance(event, MoneyWithdrawn):
            self.balance -= event.amount
        elif isinstance(event, AccountClosed):
            self.status = "closed"

    # Command methods — validate and raise events
    def create(self, owner_name: str, initial_balance: float):
        if self.status != "nonexistent":
            raise ValueError("Account already exists")
        if initial_balance < 0:
            raise ValueError("Initial balance cannot be negative")

        self.raise_event(AccountCreated(
            aggregate_id=self.id,
            version=0,
            owner_name=owner_name,
            initial_balance=initial_balance,
        ))

    def deposit(self, amount: float, description: str = ""):
        if self.status != "active":
            raise ValueError("Account is not active")
        if amount <= 0:
            raise ValueError("Deposit amount must be positive")

        self.raise_event(MoneyDeposited(
            aggregate_id=self.id,
            version=self.version + 1,
            amount=amount,
            description=description,
        ))

    def withdraw(self, amount: float, description: str = ""):
        if self.status != "active":
            raise ValueError("Account is not active")
        if amount <= 0:
            raise ValueError("Withdrawal amount must be positive")
        if self.balance < amount:
            raise ValueError("Insufficient funds")

        self.raise_event(MoneyWithdrawn(
            aggregate_id=self.id,
            version=self.version + 1,
            amount=amount,
            description=description,
        ))

    def close(self, reason: str):
        if self.status != "active":
            raise ValueError("Account is not active")
        if self.balance != 0:
            raise ValueError("Balance must be zero to close")

        self.raise_event(AccountClosed(
            aggregate_id=self.id,
            version=self.version + 1,
            reason=reason,
        ))

4. Command Handler (Write Side)

from typing import Callable

class CommandHandler:
    def __init__(self, event_store: EventStore):
        self.event_store = event_store

    def handle(self, aggregate_id: str, command: Callable[[BankAccount], None]):
        # Load aggregate from event store
        events = self.event_store.get_events(aggregate_id)
        account = BankAccount.from_events(aggregate_id, events)

        # Execute command (raises events)
        command(account)

        # Persist new events
        pending = account.get_pending_events()
        if pending:
            self.event_store.append(aggregate_id, pending, account.version - len(pending))
            account.clear_pending_events()

        return account

# Usage
store = EventStore('postgresql://user:pass@localhost/eventstore')
handler = CommandHandler(store)

account_id = str(uuid.uuid4())

# Create account
handler.handle(account_id, lambda acc: acc.create("Alice", 1000))

# Deposit
handler.handle(account_id, lambda acc: acc.deposit(500, "Salary"))

# Withdraw
handler.handle(account_id, lambda acc: acc.withdraw(200, "Groceries"))

5. Projection (Read Side)

from sqlalchemy import Column, String, Float, DateTime

class AccountProjection(Base):
    __tablename__ = 'account_projection'
    id = Column(String, primary_key=True)
    owner_name = Column(String, nullable=False)
    balance = Column(Float, nullable=False, default=0)
    status = Column(String, nullable=False, default='active')
    last_updated = Column(DateTime)

class TransactionProjection(Base):
    __tablename__ = 'transaction_projection'
    id = Column(String, primary_key=True)
    account_id = Column(String, nullable=False, index=True)
    type = Column(String, nullable=False)  # deposit, withdrawal
    amount = Column(Float, nullable=False)
    description = Column(String)
    timestamp = Column(DateTime, nullable=False)

class AccountProjector:
    """Projects events into read-optimized tables."""

    def __init__(self, session_factory):
        self.Session = session_factory

    def handle(self, event: DomainEvent):
        session = self.Session()
        try:
            # Durable dedup — an in-memory set dies on restart, a table doesn't
            result = session.execute(
                text("""INSERT INTO processed_events (event_id, processed_at)
                        VALUES (:id, NOW()) ON CONFLICT DO NOTHING"""),
                {'id': event.event_id}
            )
            if result.rowcount == 0:
                return  # already processed

            if isinstance(event, AccountCreated):
                session.execute(
                    text("""INSERT INTO account_projection (id, owner_name, balance, status, last_updated)
                            VALUES (:id, :name, :bal, 'active', :ts)"""),
                    {'id': event.aggregate_id, 'name': event.owner_name,
                     'bal': event.initial_balance, 'ts': event.timestamp}
                )

            elif isinstance(event, MoneyDeposited):
                session.execute(
                    text("""UPDATE account_projection SET balance = balance + :amt, last_updated = :ts
                            WHERE id = :id"""),
                    {'amt': event.amount, 'ts': event.timestamp, 'id': event.aggregate_id}
                )
                session.execute(
                    text("""INSERT INTO transaction_projection (id, account_id, type, amount, description, timestamp)
                            VALUES (:id, :aid, 'deposit', :amt, :desc, :ts)"""),
                    {'id': event.event_id, 'aid': event.aggregate_id,
                     'amt': event.amount, 'desc': event.description, 'ts': event.timestamp}
                )

            elif isinstance(event, MoneyWithdrawn):
                session.execute(
                    text("""UPDATE account_projection SET balance = balance - :amt, last_updated = :ts
                            WHERE id = :id"""),
                    {'amt': event.amount, 'ts': event.timestamp, 'id': event.aggregate_id}
                )
                session.execute(
                    text("""INSERT INTO transaction_projection (id, account_id, type, amount, description, timestamp)
                            VALUES (:id, :aid, 'withdrawal', :amt, :desc, :ts)"""),
                    {'id': event.event_id, 'aid': event.aggregate_id,
                     'amt': event.amount, 'desc': event.description, 'ts': event.timestamp}
                )

            elif isinstance(event, AccountClosed):
                session.execute(
                    text("""UPDATE account_projection SET status = 'closed', last_updated = :ts
                            WHERE id = :id"""),
                    {'ts': event.timestamp, 'id': event.aggregate_id}
                )

            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

The dedup row and the projection update commit in the same transaction, so a crash mid-flight either applies both or neither.

6. Query Side (Read Models)

class AccountQueryService:
    def __init__(self, session_factory):
        self.Session = session_factory

    def get_account(self, account_id: str) -> dict:
        session = self.Session()
        try:
            result = session.execute(
                text("SELECT id, owner_name, balance, status FROM account_projection WHERE id = :id"),
                {'id': account_id}
            ).fetchone()
            return dict(result._mapping) if result else None
        finally:
            session.close()

    def get_transactions(self, account_id: str, limit: int = 50) -> list:
        session = self.Session()
        try:
            results = session.execute(
                text("""SELECT type, amount, description, timestamp FROM transaction_projection
                        WHERE account_id = :id ORDER BY timestamp DESC LIMIT :lim"""),
                {'id': account_id, 'lim': limit}
            ).fetchall()
            return [dict(r._mapping) for r in results]
        finally:
            session.close()

    def get_active_accounts(self) -> list:
        session = self.Session()
        try:
            results = session.execute(
                text("SELECT id, owner_name, balance, status FROM account_projection WHERE status = 'active'")
            ).fetchall()
            return [dict(r._mapping) for r in results]
        finally:
            session.close()

7. Snapshot for Performance

class SnapshotStore:
    """Store aggregate snapshots to avoid replaying all events."""

    def __init__(self, database_url: str):
        self.engine = create_engine(database_url)
        Base.metadata.create_all(self.engine)  # creates the snapshots table too
        self.Session = sessionmaker(bind=self.engine)

    def save_snapshot(self, aggregate_id: str, version: int, state: dict):
        session = self.Session()
        try:
            session.execute(
                text("""INSERT INTO snapshots (aggregate_id, version, state, created_at)
                        VALUES (:id, :ver, :state, NOW())
                        ON CONFLICT (aggregate_id) DO UPDATE SET version = :ver, state = :state, created_at = NOW()"""),
                {'id': aggregate_id, 'ver': version, 'state': json.dumps(state)}
            )
            session.commit()
        finally:
            session.close()

    def get_snapshot(self, aggregate_id: str) -> Optional[tuple]:
        session = self.Session()
        try:
            result = session.execute(
                text("SELECT version, state FROM snapshots WHERE aggregate_id = :id"),
                {'id': aggregate_id}
            ).fetchone()
            if result:
                return result[0], json.loads(result[1])
            return None
        finally:
            session.close()

# Usage: load from snapshot, then replay only newer events
def load_aggregate(aggregate_id: str, event_store: EventStore, snapshot_store: SnapshotStore):
    snapshot = snapshot_store.get_snapshot(aggregate_id)
    if snapshot:
        version, state = snapshot
        account = BankAccount(aggregate_id)
        account.__dict__.update(state)
        # Replay only events after the snapshot
        events = event_store.get_events(aggregate_id, from_version=version)
        for event in events:
            account.apply(event)
            account.version = event.version
        return account
    else:
        events = event_store.get_events(aggregate_id)
        return BankAccount.from_events(aggregate_id, events)

How It Works

1. Event store: an append-only table storing every domain event with its aggregate ID, version, and serialized data. Events are never modified or deleted — only appended.
2. Aggregate root: holds domain logic. Command methods validate invariants and raise events; apply mutates state. Replaying the history rebuilds the aggregate at any version.
3. Optimistic concurrency: the SELECT MAX(version) check is a fast path, but the real guard is the UNIQUE(aggregate_id, version) constraint. Two concurrent writers can both pass the SELECT, but the second one fails on the constraint and raises ConcurrencyError. That check-then-insert race is why the constraint matters.
4. Projections: read-optimized tables built from events. Each projector handles the event types it cares about and commits its updates plus a processed_events row atomically, so duplicates are skipped even after a crash.
5. Snapshots: replaying 10,000 events to answer one command is slow. A snapshot stores aggregate state at a version, and loading replays only what came after.

Production Notes

• Event schema evolution: stored events are immutable, so changing an event’s shape means adding a version field and writing upcasters that transform old payloads on read. Never migrate the table in place.
• Ordering and delivery: consumers get events at least once, in order per aggregate. Global ordering isn’t guaranteed across aggregates, so don’t build logic that depends on it.
• Snapshot cadence: snapshot every N events (100 is a reasonable default) or on deploy for aggregates that accrue history fast. Measure replay time before tuning.
• GDPR and erasure: an immutable log conflicts with right-to-erasure requests. The common escape hatch is crypto-shredding — encrypt per-subject PII with a per-aggregate key and destroy the key on request, leaving the events but making them unreadable.
• Rebuilding a projection: spin up a second projection table, replay the log into it, then swap reads over. Keep the projection’s own schema version so old and new can coexist during the replay.

Variants

Event Sourcing with Kafka

# Publish events to Kafka after appending to the event store
from confluent_kafka import Producer

producer = Producer({'bootstrap.servers': 'localhost:9092'})

def append_and_publish(aggregate_id: str, events: List[DomainEvent], expected_version: int):
    event_store.append(aggregate_id, events, expected_version)
    for event in events:
        producer.produce(
            'domain-events',
            key=aggregate_id.encode('utf-8'),
            value=event.model_dump_json().encode('utf-8'),
        )
    producer.flush()

Appending to the database and publishing to Kafka isn’t atomic. A crash between the two loses or duplicates events. The transactional outbox recipe solves this by writing events to an outbox table in the same transaction and relaying them asynchronously.

Multiple Projections

# Fan out each event to every registered projector
class ProjectorManager:
    def __init__(self, projectors: list):
        self.projectors = projectors

    def handle(self, event: DomainEvent):
        for projector in self.projectors:
            projector.handle(event)

# A second projection tracking per-account activity counts
class AccountStatsProjector:
    def __init__(self, session_factory):
        self.Session = session_factory

    def handle(self, event: DomainEvent):
        session = self.Session()
        try:
            result = session.execute(
                text("""INSERT INTO processed_events (event_id, processed_at)
                        VALUES (:id, NOW()) ON CONFLICT DO NOTHING"""),
                {'id': event.event_id}
            )
            if result.rowcount == 0:
                return
            session.execute(
                text("""INSERT INTO account_stats (account_id, event_count, last_event_at)
                        VALUES (:aid, 1, :ts)
                        ON CONFLICT (account_id) DO UPDATE
                        SET event_count = account_stats.event_count + 1, last_event_at = :ts"""),
                {'aid': event.aggregate_id, 'ts': event.timestamp}
            )
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

manager = ProjectorManager([
    AccountProjector(session_factory),
    AccountStatsProjector(session_factory),
])

Each projector keeps its own dedup records and can be rebuilt independently, so a slow or broken projection doesn’t block the others.

Best Practices

• Make events immutable: never modify or delete events. A mistake gets fixed by a compensating event (MoneyRefunded), not an UPDATE. The log is the source of truth.
• Enforce concurrency in the database: a UNIQUE(aggregate_id, version) constraint is what actually prevents lost updates; application-level checks are an optimization, not a guarantee.
• Persist dedup state: the same event processed twice should produce the same result, and that must hold after a restart, so keep processed event IDs in a table, not a set().
• Snapshot long-lived aggregates: replaying thousands of events per command doesn’t scale. Snapshot on a cadence and replay the tail only.
• Separate write and read databases: the write side wants PostgreSQL for the log; the read side can be anything: Elasticsearch for search, Redis for hot lookups. Different engines per query pattern is the point of CQRS.
• Version your events: schemas drift. Add a version field and upcast old payloads on read instead of rewriting the log.

For a deeper conceptual walkthrough of how the two patterns fit together, see the CQRS + Event Sourcing combined guide.

Common Mistakes

• Storing state instead of events: the event store holds events, not the final balance. Current state is always derived; persisting it in the log breaks replay and audit.
• Check-then-insert without a constraint: SELECT MAX(version) then INSERT leaves a race window where two writers commit the same version. Without UNIQUE(aggregate_id, version) the conflict is silent.
• In-memory idempotency: a set() of processed event IDs is wiped on restart, so the first replayed event after a crash gets applied twice. Dedup has to live in the database, committed with the projection update.
• Replaying everything on every load: for long-lived aggregates that’s O(history) per command. Snapshots cut it to O(1) plus the events since the last snapshot.
• Coupling projections to the write path: a projection that fails inside the command transaction blocks writes. Projections consume events asynchronously: a dead projection retries or dead-letters while the write side keeps working.

Further Reading

• Event Sourcing — Martin Fowler: the canonical write-up on the pattern’s history and trade-offs
• Event Sourcing and CQRS patterns — microservices.io: pattern catalog entries with microservice-specific variants