Implementar Event Sourcing con CQRS en Python

Descripción general

Un auditor pregunta por qué la cuenta 42 mostraba un saldo distinto el 15 de marzo al que muestra hoy. Con una tabla CRUD ese historial ya no existe — ganó el último UPDATE. Event sourcing responde la pregunta porque cada cambio de estado se guarda como un evento inmutable en un log append-only; el saldo actual es una proyección, no el registro. CQRS encaja con él: los comandos agregan eventos en el lado de escritura, y los modelos de lectura se construyen reproduciendo eventos en tablas diseñadas para cada consulta.

Esta receta construye una implementación funcional en Python con PostgreSQL: un event store con control de concurrencia real, un aggregate root, un command handler, proyecciones con idempotencia durable y snapshots para streams largos. Para una versión del event store en MySQL o SQL Server, consulta la receta de event sourcing relacional.

Cuándo usar esto

• Sistemas que deben responder “¿cuál era el estado en el momento X?” (transacciones financieras, registros de salud, dominios con mucha regulación)
• Sistemas de lectura intensiva que necesitan varios modelos de lectura sobre los mismos eventos (vistas de lista, dashboards, índices de búsqueda)
• Dominios complejos donde reproducir eventos es la forma más rápida de depurar o probar una transición de estado
• Arquitecturas orientadas a eventos donde el log sirve además como fuente de verdad para otros consumidores

Evita event sourcing cuando:

• El dominio es CRUD simple sin requisitos de auditoría ni consultas temporales; el coste de infraestructura no te compra nada
• Nadie en el equipo puede hacerse cargo de la evolución del esquema de eventos y el mantenimiento de proyecciones; ambas son trabajo continuo, no configuración de una vez
• Los usuarios necesitan consistencia read-your-own-writes en todas partes; las proyecciones son eventualmente consistentes por diseño

Prerrequisitos

• Python 3.11+
• PostgreSQL (o cualquier base de datos con ACID)
• Paquetes sqlalchemy y pydantic

Código companion: el repo companion de event-sourcing-cqrs-pattern tiene estos ejemplos como archivos ejecutables más una demo main.py de punta a punta.

Solución

1. Definiciones de eventos

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

class DomainEvent(BaseModel):
    # default_factory se ejecuta por instancia — un default fijo se
    # evaluaría una vez al definir la clase y lo compartirían todos los eventos.
    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__ = (
        # El guardia real de concurrencia: dos writers no pueden
        # commitear el mismo par (aggregate_id, version) aunque ambos pasen el SELECT.
        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' es nombre reservado en Declarative — se mapea explícitamente
    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:
            # Chequeo rápido; el UNIQUE de arriba es la garantía real
            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()
            # Otro writer commiteó la misma versión primero
            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"

    # Métodos de comando — validar y emitir eventos
    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 (lado de escritura)

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]):
        # Cargar el aggregate desde el event store
        events = self.event_store.get_events(aggregate_id)
        account = BankAccount.from_events(aggregate_id, events)

        # Ejecutar el comando (emite eventos)
        command(account)

        # Persistir los eventos nuevos
        pending = account.get_pending_events()
        if pending:
            self.event_store.append(aggregate_id, pending, account.version - len(pending))
            account.clear_pending_events()

        return account

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

account_id = str(uuid.uuid4())

# Crear cuenta
handler.handle(account_id, lambda acc: acc.create("Alice", 1000))

# Depósito
handler.handle(account_id, lambda acc: acc.deposit(500, "Salary"))

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

5. Proyección (lado de lectura)

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:
    """Proyecta eventos en tablas optimizadas para lectura."""

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

    def handle(self, event: DomainEvent):
        session = self.Session()
        try:
            # Deduplicación durable — un set en memoria muere al reiniciar, una tabla no
            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  # ya procesado

            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()

La fila de deduplicación y la actualización de la proyección se commitean en la misma transacción, así que un fallo a mitad de camino aplica ambas o ninguna.

6. Lado de consulta (modelos de lectura)

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 para rendimiento

class SnapshotStore:
    """Guarda snapshots de aggregates para no reproducir todos los eventos."""

    def __init__(self, database_url: str):
        self.engine = create_engine(database_url)
        Base.metadata.create_all(self.engine)  # crea también la tabla snapshots
        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()

# Uso: cargar desde el snapshot y reproducir solo los eventos posteriores
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)
        # Reproducir solo los eventos posteriores al 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)

Cómo funciona

1. Event store: una tabla append-only que guarda cada evento de dominio con su aggregate ID, versión y datos serializados. Los eventos nunca se modifican ni se borran — solo se agregan.
2. Aggregate root: contiene la lógica de dominio. Los métodos de comando validan invariantes y emiten eventos; apply muta el estado. Reproducir el historial reconstruye el aggregate en cualquier versión.
3. Concurrencia optimista: el SELECT MAX(version) es un camino rápido, pero el guardia real es el constraint UNIQUE(aggregate_id, version). Dos writers concurrentes pueden pasar el SELECT, pero el segundo falla en el constraint y lanza ConcurrencyError. Esa ventana de carrera entre chequeo e inserción es la razón del constraint.
4. Proyecciones: tablas optimizadas para lectura construidas desde eventos. Cada proyector maneja los tipos de evento que le interesan y commitea sus actualizaciones junto a una fila en processed_events de forma atómica, así los duplicados se saltan incluso tras un fallo.
5. Snapshots: reproducir 10.000 eventos para atender un comando es lento. Un snapshot guarda el estado del aggregate en una versión, y la carga reproduce solo lo que vino después.

Notas de producción

• Evolución del esquema de eventos: los eventos almacenados son inmutables, así que cambiar la forma de un evento implica añadir un campo de versión y escribir upcasters que transformen los payloads viejos al leer. Nunca migres la tabla en el lugar.
• Orden y entrega: los consumidores reciben eventos al menos una vez, en orden por aggregate. No hay orden global garantizado entre aggregates, así que no construyas lógica que dependa de ello.
• Cadencia de snapshots: haz snapshot cada N eventos (100 es un default razonable) o en cada despliegue para aggregates que acumulan historial rápido. Mide el tiempo de replay antes de ajustar.
• GDPR y borrado: un log inmutable choca con las solicitudes de derecho al olvido. La salida habitual es el crypto-shredding — cifrar los datos personales por sujeto con una clave por aggregate y destruir la clave ante la solicitud, dejando los eventos pero haciéndolos ilegibles.
• Reconstruir una proyección: levanta una segunda tabla de proyección, reproduce el log dentro y luego cambia las lecturas. Mantén una versión del esquema de la proyección para que la vieja y la nueva coexistan durante el replay.

Variantes

Event Sourcing con Kafka

# Publicar eventos a Kafka después de agregarlos al 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()

Escribir en la base de datos y publicar en Kafka no es atómico. Un fallo entre ambas operaciones pierde o duplica eventos. La receta de transactional outbox lo resuelve escribiendo los eventos en una tabla outbox dentro de la misma transacción y reenviándolos de forma asíncrona.

Múltiples proyecciones

# Repartir cada evento a todos los proyectores registrados
class ProjectorManager:
    def __init__(self, projectors: list):
        self.projectors = projectors

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

# Una segunda proyección que cuenta la actividad por cuenta
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),
])

Cada proyector lleva sus propios registros de deduplicación y puede reconstruirse de forma independiente, así que una proyección lenta o rota no bloquea a las demás.

Mejores prácticas

• Hacer los eventos inmutables: nunca modifiques ni borres eventos. Un error se corrige con un evento compensatorio (MoneyRefunded), no con un UPDATE. El log es la fuente de verdad.
• Imponer la concurrencia en la base de datos: el constraint UNIQUE(aggregate_id, version) es lo que realmente evita lost updates; los chequeos a nivel de aplicación son una optimización, no una garantía.
• Persistir el estado de deduplicación: el mismo evento procesado dos veces debe dar el mismo resultado, y eso tiene que cumplirse tras un reinicio, así que guarda los event IDs procesados en una tabla, no en un set().
• Snapshot de aggregates de larga duración: reproducir miles de eventos por comando no escala. Haz snapshot con una cadencia fija y reproduce solo la cola.
• Separar las bases de escritura y lectura: el lado de escritura quiere PostgreSQL para el log; el de lectura puede ser cualquier cosa: Elasticsearch para búsqueda, Redis para lookups calientes. Motores distintos por patrón de consulta es justamente el punto de CQRS.
• Versionar los eventos: los esquemas cambian. Añade un campo de versión y transforma los payloads viejos al leer en vez de reescribir el log.

Para un recorrido conceptual más profundo de cómo encajan los dos patrones, consulta la guía combinada de CQRS + Event Sourcing.

Errores comunes

• Guardar estado en lugar de eventos: el event store contiene eventos, no el saldo final. El estado actual siempre se deriva; persistirlo en el log rompe el replay y la auditoría.
• Chequear e insertar sin constraint: SELECT MAX(version) y luego INSERT deja una ventana de carrera donde dos writers commitean la misma versión. Sin UNIQUE(aggregate_id, version) el conflicto es silencioso.
• Idempotencia en memoria: un set() de event IDs procesados se borra al reiniciar, así que el primer evento reproducido tras un fallo se aplica dos veces. La deduplicación tiene que vivir en la base de datos, commiteada con la actualización de la proyección.
• Reproducir todo en cada carga: para aggregates de larga duración eso es O(historial) por comando. Los snapshots lo reducen a O(1) más los eventos desde el último snapshot.
• Acoplar las proyecciones al camino de escritura: una proyección que falla dentro de la transacción del comando bloquea las escrituras. Las proyecciones consumen eventos de forma asíncrona: una proyección muerta reintenta o va a dead-letter mientras el lado de escritura sigue funcionando.

Lecturas adicionales

• Event Sourcing — Martin Fowler: el texto canónico sobre la historia del patrón y sus trade-offs
• Patrones Event Sourcing y CQRS — microservices.io: entradas del catálogo de patrones con variantes para microservicios