What Is Event Sourcing? Event Stores, Replay, and Practical Examples
Event Sourcing is a software architecture pattern in which application state is persisted as a sequence of events rather than storing only its latest value. Every event represents a meaningful change that has already occurred, such as an account being opened, money being deposited, or an order being cancelled. The current state of an aggregate can be reconstructed by replaying its events in order.
Event Sourcing is often associated with Command Query Responsibility Segregation (CQRS) and event-driven architecture, but these patterns solve different problems. Event Sourcing determines how authoritative state is stored; CQRS separates write and read models; event-driven architecture defines how components communicate through events. This guide explains event stores, aggregate reconstruction, projections, snapshots, event versioning, and production failure handling, followed by a Python, FastAPI, PostgreSQL, and SQLAlchemy implementation of an event-sourced account ledger.
Table of Contents
- What Is Event Sourcing?
- How Event Sourcing Works
- Designing a Production Event Store
- Production Design Example: Account Ledger
- Ready-to-Use Example with FastAPI and PostgreSQL
- Event Sourcing vs. CQRS and Event-Driven Architecture
- Common Mistakes and Production Checklist
- Frequently Asked Questions
- Conclusion
What Is Event Sourcing?
Event Sourcing stores the sequence of state changes that occurred within an aggregate and uses those events to determine the aggregate's current state. Instead of persisting only that an account has a balance of 375 dollars, the system can store the operations that produced that balance.
This changes the persistence model significantly. The event history becomes authoritative, while current-state tables, dashboards, and search documents can be derived from that history.
Current-State Persistence vs. Event Sourcing
In a conventional CRUD application, a row generally represents the latest state:
-- Conventional current-state persistence.
SELECT account_id, balance_cents
FROM accounts
WHERE account_id = 'account-1001';
-- Example result:
-- account-1001 | 37500The database tells us what the balance is, but not necessarily how it arrived at that value. A separate audit or transaction table may provide additional history.
With Event Sourcing, the authoritative state is an ordered event stream:
[
{
"version": 1,
"event_type": "AccountOpened",
"data": {"initial_balance_cents": 0}
},
{
"version": 2,
"event_type": "MoneyDeposited",
"data": {"amount_cents": 50000}
},
{
"version": 3,
"event_type": "MoneyWithdrawn",
"data": {"amount_cents": 12500}
}
]Replaying the events produces a balance of 37,500 cents. The latest state can be calculated whenever needed or maintained in a separate projection for efficient reads.
The important difference is that the event stream is the source of truth. The balance projection is derived state that can be reconstructed from the stored events.
Core Principles and Terminology
Event-sourced systems usually involve the following concepts:
- Domain event: An immutable fact representing something that happened in the domain.
- Event stream: An ordered sequence of events belonging to an aggregate.
- Aggregate: A business consistency boundary responsible for validating state transitions.
- Event store: A persistence system that durably stores events.
- Replay: Reconstructing state by applying historical events in sequence.
- Projection: A derived representation built from events, commonly optimized for queries.
- Snapshot: A stored checkpoint of aggregate state used to avoid replaying the entire history.
- Stream version: An ordering and concurrency-control value associated with an aggregate's events.
Events should describe completed facts, usually using past-tense business names such as AccountOpened, PaymentCaptured, or OrderCancelled.
A command such as WithdrawMoney requests a change and may be rejected. A MoneyWithdrawn event records a change that has already been accepted.
How Event Sourcing Works
Event-sourced applications generally accept commands, load an aggregate's history, reconstruct its state, validate the requested operation, and append new events.
Read models can be updated as part of the same transaction or asynchronously through event processors, depending on the system's consistency and scalability requirements.
Events as the Source of Truth
In a traditional application, the authoritative record might be the current account balance. With Event Sourcing, the authoritative records are the accepted events that changed the balance.
This enables several operations that are difficult when only the latest state is stored:
- Reconstruct the state of an aggregate at a previous version.
- Explain which accepted changes produced the current state.
- Build new read models from existing history.
- Diagnose unexpected state transitions.
- Reprocess events after a projection bug has been fixed.
However, event history must be interpreted using the correct historical business semantics. Replaying old events through new logic that changes their meaning can produce incorrect results.
For financial systems, an event-sourced demonstration is not automatically a complete accounting ledger. Real accounting requires additional controls such as balanced journal entries, currency rules, reconciliation, and authorization.
Aggregate Reconstruction and Replay
Reconstruction applies each event to an initially empty aggregate state.
The order matters: a withdrawal cannot be applied before the account exists, and an event referring to a later version should not silently skip missing history.
from dataclasses import dataclass
from typing import Any
@dataclass
class AccountState:
is_open: bool = False
balance_cents: int = 0
version: int = 0
def replay_account(
events: list[dict[str, Any]],
) -> AccountState:
state = AccountState()
for event in events:
expected = state.version + 1
if event["version"] != expected:
raise ValueError("Invalid event sequence")
event_type = event["event_type"]
data = event["data"]
if event_type == "AccountOpened":
if state.is_open:
raise ValueError("Account already opened")
state.is_open = True
elif event_type == "MoneyDeposited":
if not state.is_open:
raise ValueError("Account does not exist")
state.balance_cents += data["amount_cents"]
elif event_type == "MoneyWithdrawn":
if not state.is_open:
raise ValueError("Account does not exist")
state.balance_cents -= data["amount_cents"]
else:
raise ValueError(
f"Unknown event type: {event_type}"
)
state.version = expected
return stateThis simplified replay function assumes stored events were validated before being accepted. A production aggregate should also validate event payloads and the validity of historical transitions, or use schema-specific event decoders with equivalent guarantees.
Replay must not send emails, charge payment providers, or invoke other external side effects. Reconstructing historical state should be deterministic and independent of external systems.
Event Streams, Ordering, and Versions
Events are normally ordered within an aggregate stream rather than across the entire system.
For an account identified by account-1001, the stream might have versions 1, 2, 3, and 4.
Another account can have its own independent sequence.
Per-aggregate versions are valuable for detecting concurrency conflicts. Two command handlers may load version 3 and independently attempt to append version 4.
A database uniqueness constraint on (aggregate_id, version) can ensure only one conflicting append succeeds. The other command must reload the aggregate and reevaluate its business rules rather than blindly appending stale results.
Projections and Read Models
Replaying hundreds or thousands of events for every API read would be inefficient. A projection provides a current-state representation derived from the event stream.
For example, an account projection can contain:
{
"account_id": "account-1001",
"balance_cents": 37500,
"version": 3
}The projection supports fast account-balance queries while the event store retains the authoritative history.
Projections can be updated synchronously in the same database transaction as new events or asynchronously by event consumers.
Synchronous projections can provide immediately consistent reads when the event store and projection share one transaction. Asynchronous projections support greater separation but introduce lag and eventual consistency.
These are distinct choices. Event Sourcing does not require a separate database, message broker, or asynchronous projection worker.
Designing a Production Event Store
Event storage needs stronger contracts than an ordinary append-only log. Events are authoritative business history, so ordering, durability, schema evolution, and concurrency handling must be explicit.
The event store also needs operational processes for recovering projections, investigating corrupted streams, and safely replaying historical events.
Event Schema and Metadata
A useful event envelope includes identity, aggregate information, ordering, type, schema version, timestamp, and business payload.
| Field | Purpose |
|---|---|
event_id |
Identifies an individual event |
aggregate_id |
Identifies the stream's business entity |
version |
Orders events within the aggregate |
event_type |
Names the completed business fact |
schema_version |
Identifies the event payload format |
occurred_at |
Records when the event occurred |
data |
Contains the business event payload |
Production systems commonly add correlation IDs, causation IDs, tenant identifiers, and trace metadata.
Use stable business-oriented event contracts rather than serializing internal ORM entities. Otherwise, database refactoring can accidentally alter the meaning or compatibility of historical events.
Optimistic Concurrency and Atomic Appends
An event append should be conditional on the version of the stream previously loaded by the command handler.
For example, a handler that read version 4 should only be allowed to append version 5 if the stream has not changed.
A unique database constraint on aggregate ID and version provides one mechanism for detecting conflicting appends.
For systems that also maintain a current-state projection, the projection row can be locked before replaying and modifying the aggregate. That lock serializes commands for the same aggregate while allowing unrelated aggregates to proceed concurrently.
This locking strategy is used in the implementation below. It keeps the educational system straightforward while relying on PostgreSQL's transactional behavior.
Snapshots and Replay Performance
As event streams grow, reconstructing an aggregate from its first event can become expensive.
A snapshot stores the aggregate's state at a particular stream version. Later reconstruction loads the snapshot and replays only events with higher versions.
For example, a snapshot at version 1,000 allows a handler to apply versions 1,001 through 1,025 instead of replaying the entire stream.
Snapshot creation should be an optimization rather than a second independently authoritative state source.
A snapshot must record the version it represents, and its creation should not make historical event data unrecoverable.
Whether snapshots are needed depends on measured replay time, event volume, and aggregate complexity. Avoid introducing snapshot infrastructure before it provides a meaningful benefit.
Event Versioning and Schema Evolution
Stored events may remain in the database for years. Unlike an ordinary API payload, historical events cannot simply be assumed to use the latest schema.
For example, an older MoneyDeposited event may contain only an amount. A newer version might also contain a currency and source reference.
Applications need a defined strategy for interpreting both schemas.
Common approaches include:
- Supporting multiple event schema versions in readers.
- Upcasting older event payloads into the current in-memory representation.
- Maintaining backward-compatible additions where possible.
- Testing replay against historical event fixtures.
- Validating projection rebuilds after event-handler changes.
A business policy change should not silently change the meaning of historical events. For example, a historical withdrawal should not be revalidated using today's account limits and rejected during reconstruction.
Production Design Example: Account Ledger
Consider a small account application that supports opening accounts, depositing funds, withdrawing funds, and querying account balances.
This is an educational ledger-like example rather than a complete banking or accounting implementation. The goal is to demonstrate durable event history, aggregate reconstruction, atomic append behavior, and a current-state read model.
Business Requirements and Aggregate Boundaries
The application enforces the following rules:
- An account must be opened before it can accept deposits or withdrawals.
- Account IDs are unique.
- Deposits and withdrawals must use positive integer cents.
- A withdrawal cannot exceed the available balance.
- Events for one account must have consecutive versions.
- Committed events and the account's read projection must remain consistent.
- Commands for the same account must not overwrite one another.
Each account is an aggregate with its own stream. A deposit affects only its account, so commands involving different accounts can normally execute independently.
The read model contains the current balance and stream version, while the event stream records the accepted changes that produced those values.
Command Processing and Projection Updates
The command-processing workflow is:
- Start a PostgreSQL transaction.
- Lock the account's projection row for modification.
- Load and replay its event stream.
- Validate the requested command against reconstructed state.
- Append the new event at the next stream version.
- Update the account projection in the same transaction.
- Commit the transaction.
The lock prevents two balance-changing commands from independently acting on the same stale account state.
For account creation, the primary key and stream-version uniqueness constraints protect against concurrent duplicate openings.
The example uses a synchronous projection because it makes transactional consistency easy to demonstrate. Larger systems can move projections to independent consumers when that trade-off is justified.
Concurrency, Failure Recovery, and Idempotency
If a command fails business validation, no new event should be committed.
If a database operation fails, the transaction must roll back both the event append and any projection update.
Retries require special care. If a client times out after a transaction commits, it may not know whether the operation succeeded. Blindly retrying a deposit could create a second accepted event.
A production command API should support stable idempotency keys or command identifiers, enforced atomically in persistent storage. Repeating an already committed command should return the original outcome rather than applying the business operation again.
The minimal implementation below provides concurrency protection but does not implement the complete idempotency-key lifecycle. This is an important production extension, especially for financial operations.
Ready-to-Use Example with FastAPI and PostgreSQL
The implementation uses Python 3.12 or newer, FastAPI, SQLAlchemy 2, PostgreSQL, and pytest.
It stores events as the authoritative history and maintains an account projection in the same transaction. Domain replay is independent of FastAPI and SQLAlchemy.
Project Structure and Dependencies
Create the following structure and add empty __init__.py files inside Python package directories:
event-sourced-accounts/
app/
__init__.py
domain.py
persistence.py
service.py
api.py
tests/
test_domain.py
requirements.txt
docker-compose.ymlCreate requirements.txt:
fastapi>=0.115,<1
uvicorn>=0.30,<1
sqlalchemy>=2.0,<3
psycopg[binary]>=3.2,<4
pytest>=8,<10
httpx>=0.27,<1For production, pin tested dependency versions and use versioned migrations rather than creating tables dynamically at startup.
Domain Events and Account Aggregate
The domain layer defines how accepted events change the state of an account. Commands are validated against reconstructed state before producing new events.
Create app/domain.py:
from dataclasses import dataclass
from typing import Any
class AccountNotFound(Exception):
pass
class AccountAlreadyExists(Exception):
pass
class InsufficientFunds(Exception):
pass
@dataclass(frozen=True)
class AccountEvent:
version: int
event_type: str
data: dict[str, Any]
@dataclass
class Account:
is_open: bool = False
balance_cents: int = 0
version: int = 0
def apply(self, event: AccountEvent) -> None:
if event.version != self.version + 1:
raise ValueError("Nonconsecutive event version")
if event.event_type == "AccountOpened":
if self.is_open:
raise ValueError("Account already opened")
self.is_open = True
elif event.event_type == "MoneyDeposited":
if not self.is_open:
raise ValueError("Account not opened")
amount = event.data["amount_cents"]
self._validate_amount(amount)
self.balance_cents += amount
elif event.event_type == "MoneyWithdrawn":
if not self.is_open:
raise ValueError("Account not opened")
amount = event.data["amount_cents"]
self._validate_amount(amount)
if amount > self.balance_cents:
raise ValueError(
"Historical withdrawal exceeds balance"
)
self.balance_cents -= amount
else:
raise ValueError(
f"Unknown event type: {event.event_type}"
)
self.version = event.version
@staticmethod
def _validate_amount(amount: int) -> None:
if type(amount) is not int or amount <= 0:
raise ValueError(
"Amount must be a positive integer"
)
def decide_deposit(
self,
amount: int,
) -> AccountEvent:
if not self.is_open:
raise AccountNotFound()
self._validate_amount(amount)
return AccountEvent(
version=self.version + 1,
event_type="MoneyDeposited",
data={"amount_cents": amount},
)
def decide_withdrawal(
self,
amount: int,
) -> AccountEvent:
if not self.is_open:
raise AccountNotFound()
self._validate_amount(amount)
if amount > self.balance_cents:
raise InsufficientFunds()
return AccountEvent(
version=self.version + 1,
event_type="MoneyWithdrawn",
data={"amount_cents": amount},
)
def replay(events: list[AccountEvent]) -> Account:
account = Account()
for event in events:
account.apply(event)
return accountTwo different operations are visible: decide_* validates a new command and produces a proposed event, while apply() reconstructs state from previously accepted events.
Replay does not call payment gateways or send messages. It only interprets persisted business facts.
The design intentionally uses a single supported currency, integer cents, and a nonnegative balance. A real financial domain requires richer accounting invariants and authorization rules.
Event Store and Account Read Model
The database contains two tables: an append-only event stream and an account projection optimized for reading balances.
Create app/persistence.py:
import os
from datetime import datetime, timezone
from sqlalchemy import (
DateTime,
ForeignKey,
Integer,
JSON,
String,
UniqueConstraint,
create_engine,
select,
)
from sqlalchemy.orm import (
DeclarativeBase,
Mapped,
Session,
mapped_column,
sessionmaker,
)
from app.domain import AccountEvent
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+psycopg://app:app@localhost:5432/accounts",
)
engine = create_engine(
DATABASE_URL,
pool_pre_ping=True,
)
SessionFactory = sessionmaker(
bind=engine,
expire_on_commit=False,
)
class Base(DeclarativeBase):
pass
class AccountViewRow(Base):
__tablename__ = "account_views"
account_id: Mapped[str] = mapped_column(
String(100),
primary_key=True,
)
balance_cents: Mapped[int] = mapped_column(
Integer,
nullable=False,
default=0,
)
version: Mapped[int] = mapped_column(
Integer,
nullable=False,
default=0,
)
class EventRow(Base):
__tablename__ = "account_events"
__table_args__ = (
UniqueConstraint(
"account_id",
"version",
name="uq_account_event_version",
),
)
event_id: Mapped[str] = mapped_column(
String(36),
primary_key=True,
)
account_id: Mapped[str] = mapped_column(
String(100),
ForeignKey("account_views.account_id"),
nullable=False,
index=True,
)
version: Mapped[int] = mapped_column(
Integer,
nullable=False,
)
event_type: Mapped[str] = mapped_column(
String(100),
nullable=False,
)
schema_version: Mapped[int] = mapped_column(
Integer,
nullable=False,
default=1,
)
data: Mapped[dict] = mapped_column(
JSON,
nullable=False,
)
occurred_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
)
def load_events(
session: Session,
account_id: str,
) -> list[AccountEvent]:
statement = (
select(EventRow)
.where(EventRow.account_id == account_id)
.order_by(EventRow.version)
)
rows = session.scalars(statement).all()
return [
AccountEvent(
version=row.version,
event_type=row.event_type,
data=row.data,
)
for row in rows
]The uniqueness constraint on (account_id, version) ensures a stream cannot contain two events with the same version.
The foreign key ensures events belong to an existing account projection row. The read projection acts as a convenient transactional locking record in this implementation.
Because the authoritative history is the event stream, a production system should also protect it against unauthorized modifications and define operational controls for detecting accidental event edits or deletes.
Command Handlers and Transaction Boundaries
The application service coordinates domain replay, database locking, new event persistence, and projection updates.
Create app/service.py:
from datetime import datetime, timezone
from uuid import uuid4
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from app.domain import (
Account,
AccountAlreadyExists,
AccountEvent,
AccountNotFound,
replay,
)
from app.persistence import (
AccountViewRow,
EventRow,
SessionFactory,
load_events,
)
class AccountService:
def open_account(self, account_id: str) -> dict:
try:
with SessionFactory.begin() as session:
if session.get(
AccountViewRow,
account_id,
) is not None:
raise AccountAlreadyExists()
session.add(
AccountViewRow(
account_id=account_id,
balance_cents=0,
version=1,
)
)
event = AccountEvent(
version=1,
event_type="AccountOpened",
data={},
)
session.add(
self._event_row(account_id, event)
)
except IntegrityError as exc:
# Also handles concurrent openings of
# the same account identifier.
raise AccountAlreadyExists() from exc
return {
"account_id": account_id,
"balance_cents": 0,
"version": 1,
}
def deposit(
self,
account_id: str,
amount_cents: int,
) -> dict:
return self._change_balance(
account_id,
amount_cents,
operation="deposit",
)
def withdraw(
self,
account_id: str,
amount_cents: int,
) -> dict:
return self._change_balance(
account_id,
amount_cents,
operation="withdraw",
)
def _change_balance(
self,
account_id: str,
amount_cents: int,
operation: str,
) -> dict:
with SessionFactory.begin() as session:
# Serialize commands for this account.
view = session.scalar(
select(AccountViewRow)
.where(
AccountViewRow.account_id == account_id
)
.with_for_update()
)
if view is None:
raise AccountNotFound()
# Reconstruct authoritative domain state.
account = replay(
load_events(session, account_id)
)
if account.version != view.version:
raise RuntimeError(
"Projection version mismatch"
)
if operation == "deposit":
event = account.decide_deposit(
amount_cents
)
elif operation == "withdraw":
event = account.decide_withdrawal(
amount_cents
)
else:
raise ValueError("Unknown operation")
# Apply the accepted event to derive
# the new projection state.
account.apply(event)
session.add(
self._event_row(account_id, event)
)
# Same transaction as the event append.
view.balance_cents = account.balance_cents
view.version = account.version
result = {
"account_id": account_id,
"balance_cents": account.balance_cents,
"version": account.version,
}
return result
def get_account(self, account_id: str) -> dict:
with SessionFactory() as session:
row = session.get(
AccountViewRow,
account_id,
)
if row is None:
raise AccountNotFound()
return {
"account_id": row.account_id,
"balance_cents": row.balance_cents,
"version": row.version,
}
def get_history(self, account_id: str) -> list[dict]:
with SessionFactory() as session:
if session.get(
AccountViewRow,
account_id,
) is None:
raise AccountNotFound()
rows = session.scalars(
select(EventRow)
.where(EventRow.account_id == account_id)
.order_by(EventRow.version)
).all()
return [
{
"event_id": row.event_id,
"version": row.version,
"event_type": row.event_type,
"schema_version": row.schema_version,
"data": row.data,
"occurred_at": row.occurred_at,
}
for row in rows
]
@staticmethod
def _event_row(
account_id: str,
event: AccountEvent,
) -> EventRow:
return EventRow(
event_id=str(uuid4()),
account_id=account_id,
version=event.version,
event_type=event.event_type,
schema_version=1,
data=event.data,
occurred_at=datetime.now(timezone.utc),
)The most important part is the transaction around _change_balance(). It locks the account projection, reconstructs the account from events, validates the operation, appends a new event, and updates the projection before committing.
This design provides atomic local consistency between the event stream and its current-state read model.
For high-volume event streams, replaying every historical event on every command becomes inefficient. A snapshot mechanism can be added later without changing the meaning of the authoritative event history.
In a larger application, event store access, aggregate decisions, and projection persistence should be separated into interfaces and adapters. They are kept in one service here to make the transaction and replay sequence easy to understand.
FastAPI Endpoints and Event History
The HTTP adapter exposes account creation, deposits, withdrawals, balance queries, and event history.
Create app/api.py:
from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from app.domain import (
AccountAlreadyExists,
AccountNotFound,
InsufficientFunds,
)
from app.persistence import Base, engine
from app.service import AccountService
class AccountRequest(BaseModel):
account_id: str = Field(min_length=1)
class AmountRequest(BaseModel):
amount_cents: int = Field(gt=0)
class AccountResponse(BaseModel):
account_id: str
balance_cents: int
version: int
class EventResponse(BaseModel):
event_id: str
version: int
event_type: str
schema_version: int
data: dict
occurred_at: datetime
@asynccontextmanager
async def lifespan(app: FastAPI):
# Local demo only. Use migrations in production.
Base.metadata.create_all(bind=engine)
yield
engine.dispose()
app = FastAPI(
title="Event-Sourced Accounts",
lifespan=lifespan,
)
service = AccountService()
@app.post(
"/accounts",
response_model=AccountResponse,
status_code=201,
)
def open_account(request: AccountRequest):
try:
return service.open_account(
request.account_id
)
except AccountAlreadyExists as exc:
raise HTTPException(
status_code=409,
detail="Account already exists",
) from exc
@app.post(
"/accounts/{account_id}/deposits",
response_model=AccountResponse,
)
def deposit(account_id: str, request: AmountRequest):
try:
return service.deposit(
account_id,
request.amount_cents,
)
except AccountNotFound as exc:
raise HTTPException(
status_code=404,
detail="Account not found",
) from exc
@app.post(
"/accounts/{account_id}/withdrawals",
response_model=AccountResponse,
)
def withdraw(account_id: str, request: AmountRequest):
try:
return service.withdraw(
account_id,
request.amount_cents,
)
except AccountNotFound as exc:
raise HTTPException(
status_code=404,
detail="Account not found",
) from exc
except InsufficientFunds as exc:
raise HTTPException(
status_code=409,
detail="Insufficient funds",
) from exc
@app.get(
"/accounts/{account_id}",
response_model=AccountResponse,
)
def get_account(account_id: str):
try:
return service.get_account(account_id)
except AccountNotFound as exc:
raise HTTPException(
status_code=404,
detail="Account not found",
) from exc
@app.get(
"/accounts/{account_id}/events",
response_model=list[EventResponse],
)
def get_events(account_id: str):
try:
return service.get_history(account_id)
except AccountNotFound as exc:
raise HTTPException(
status_code=404,
detail="Account not found",
) from excThe read endpoint uses the projection rather than replaying the full history. The event-history endpoint retrieves the ordered event stream for inspection.
These endpoints intentionally omit authentication, authorization, rate limiting, and idempotency keys. They are suitable for demonstrating the architecture locally, not for processing real financial transactions.
Unit Tests for Replay and Business Rules
Event-sourced systems need tests that validate aggregate decisions and confirm that historical event streams reconstruct correct state.
Create tests/test_domain.py:
import pytest
from app.domain import (
AccountEvent,
InsufficientFunds,
replay,
)
def example_events() -> list[AccountEvent]:
return [
AccountEvent(
version=1,
event_type="AccountOpened",
data={},
),
AccountEvent(
version=2,
event_type="MoneyDeposited",
data={"amount_cents": 50000},
),
AccountEvent(
version=3,
event_type="MoneyWithdrawn",
data={"amount_cents": 12500},
),
]
def test_replay_restores_balance():
account = replay(example_events())
assert account.is_open
assert account.balance_cents == 37500
assert account.version == 3
def test_deposit_produces_next_event():
account = replay(example_events())
event = account.decide_deposit(2500)
assert event.version == 4
assert event.event_type == "MoneyDeposited"
account.apply(event)
assert account.balance_cents == 40000
assert account.version == 4
def test_withdrawal_rejected_when_insufficient():
account = replay(example_events())
with pytest.raises(InsufficientFunds):
account.decide_withdrawal(40000)
def test_nonconsecutive_event_rejected():
events = [
AccountEvent(
version=1,
event_type="AccountOpened",
data={},
),
AccountEvent(
version=3,
event_type="MoneyDeposited",
data={"amount_cents": 1000},
),
]
with pytest.raises(ValueError):
replay(events)
def test_invalid_historical_event_rejected():
events = [
AccountEvent(
version=1,
event_type="AccountOpened",
data={},
),
AccountEvent(
version=2,
event_type="MoneyWithdrawn",
data={"amount_cents": 1000},
),
]
with pytest.raises(ValueError):
replay(events)
def test_replay_is_deterministic():
first = replay(example_events())
second = replay(example_events())
assert first.balance_cents == second.balance_cents
assert first.version == second.version
def test_invalid_amount_rejected():
account = replay(example_events())
with pytest.raises(ValueError):
account.decide_deposit(0)
with pytest.raises(ValueError):
account.decide_withdrawal(-1)These tests verify event ordering, business invariants, state reconstruction, and deterministic replay.
Additional PostgreSQL integration tests should cover concurrent withdrawals, duplicate account creation, event/version uniqueness, and failure rollback when projection updates or event inserts fail.
For production systems, maintain historical event fixtures from previous schema versions and run replay tests against them before deploying new event handlers.
Run and Verify the Application
Use Docker Compose to start PostgreSQL locally.
Create docker-compose.yml:
services:
postgres:
image: postgres:16-alpine
environment:
POSTGRES_DB: accounts
POSTGRES_USER: app
POSTGRES_PASSWORD: app
ports:
- "127.0.0.1:5432:5432"
healthcheck:
test:
- CMD-SHELL
- pg_isready -U app -d accounts
interval: 5s
timeout: 3s
retries: 10Run the application and test the event history:
# Start PostgreSQL.
docker compose up -d --wait
# Create a Python environment.
python3 -m venv .venv
source .venv/bin/activate
# Install dependencies.
pip install -r requirements.txt
# Run domain tests.
pytest -q
# Configure PostgreSQL.
export DATABASE_URL="postgresql+psycopg://app:app@localhost:5432/accounts"
# Start FastAPI in a terminal.
uvicorn app.api:app --reload
# In another terminal, open an account.
curl -i -X POST \
http://127.0.0.1:8000/accounts \
-H "Content-Type: application/json" \
-d '{"account_id": "account-1001"}'
# Deposit 500 dollars.
curl -i -X POST \
http://127.0.0.1:8000/accounts/account-1001/deposits \
-H "Content-Type: application/json" \
-d '{"amount_cents": 50000}'
# Withdraw 125 dollars.
curl -i -X POST \
http://127.0.0.1:8000/accounts/account-1001/withdrawals \
-H "Content-Type: application/json" \
-d '{"amount_cents": 12500}'
# Read the current balance projection.
curl -i \
http://127.0.0.1:8000/accounts/account-1001
# Read the complete ordered event history.
curl -i \
http://127.0.0.1:8000/accounts/account-1001/events
# Inspect the event stream directly.
docker compose exec postgres \
psql -U app -d accounts \
-c 'SELECT account_id, version, event_type, data FROM account_events ORDER BY account_id, version;'
# Inspect the read projection.
docker compose exec postgres \
psql -U app -d accounts \
-c 'SELECT account_id, balance_cents, version FROM account_views;'
# Remove the local database when finished.
docker compose downFor a fresh database, the account should end with a balance of 37,500 cents and stream version 3.
The event store should contain AccountOpened, MoneyDeposited, and MoneyWithdrawn events. The account projection should contain the same latest version and the derived balance.
If a withdrawal exceeds the available balance, the API should return 409 Conflict without committing another event.
Event Sourcing vs. CQRS and Event-Driven Architecture
Event Sourcing, CQRS, and event-driven architecture frequently appear together, but they solve different problems and can be used independently.
Understanding these distinctions prevents projects from adopting unnecessary message brokers, dual databases, or event processing frameworks.
Pattern Differences
| Pattern | Main Purpose | Requires Event Store? |
|---|---|---|
| Event Sourcing | Persist authoritative state as historical events | Yes, or equivalent durable event history |
| CQRS | Separate command and query models | No |
| Event-Driven Architecture | Communicate through event publication and consumption | No |
| Transactional Outbox | Atomically record business changes and outgoing messages | No |
A CQRS application may use a conventional PostgreSQL table as its source of truth and a separate read projection for queries.
An event-driven application may publish OrderCreated after inserting a normal order record without storing its complete history as events.
An Event Sourcing application may store events and maintain projections inside a single PostgreSQL database, as demonstrated above.
For a deeper introduction to separating write and read models, see What Is CQRS?.
Benefits, Costs, and When to Use Event Sourcing
Event Sourcing is useful when historical transitions are fundamental to the application rather than incidental audit information.
Potential benefits include:
- Reconstructing historical aggregate state.
- Explaining how current state was reached.
- Building new projections from existing events.
- Supporting detailed domain-level traceability.
- Separating authoritative write history from optimized read models.
- Enabling controlled replay after projection implementation changes.
These benefits have real costs:
- Historical event formats require long-term compatibility.
- Event histories can become large and expensive to replay.
- Replay must avoid external side effects.
- Data correction and deletion require deliberate policies.
- Concurrency and event ordering must be enforced.
- Projection recovery and lag monitoring add operational work.
- Debugging may require interpreting long sequences of historical events.
Event Sourcing is often appropriate for domains where business history itself matters, such as financial operations, complex workflows, or state transitions with significant audit requirements.
It is usually unnecessary for a small CRUD application whose main requirement is to store and retrieve the latest record.
Common Mistakes and Production Checklist
The difficulty of Event Sourcing is not primarily inserting events into a table. It is maintaining reliable and interpretable history over years of application changes.
Production implementations require careful attention to historical compatibility, concurrency, recovery, security, and operational visibility.
Common Implementation Mistakes
Common mistakes include:
- Treating a message broker as the source of truth automatically: Transport retention and persistence requirements may not match event-store guarantees.
- Confusing events with commands: Commands request actions; events record completed facts.
- Publishing events instead of storing them atomically: External delivery does not replace durable aggregate history.
- Ignoring aggregate versions: Concurrent writers can create inconsistent or conflicting histories.
- Replaying external side effects: Historical reconstruction can accidentally repeat payments, emails, or shipments.
- Changing event meanings: Older data becomes incompatible with the current application.
- Deleting or editing historical events casually: Reconstructed state and audit integrity can be compromised.
- Assuming projections are always current: Asynchronous consumers may lag behind committed events.
- Ignoring retry idempotency: Retried commands may produce duplicate business effects.
- Skipping snapshot planning: Very long streams can make command processing slow.
- Assuming event history equals compliant accounting: Domain-specific ledger and reconciliation controls are still required.
- Using Event Sourcing for simple CRUD by default: Operational complexity may outweigh the benefits.
A good design also distinguishes event-sourced state from integration messages. Internal events may contain details that should not be exposed to external subscribers.
Production Checklist
Before deploying an event-sourced application, verify that:
- Every aggregate has a stable stream identifier.
- Events use meaningful business names and immutable payloads.
- Stream versions are consecutive and protected by database constraints.
- Concurrent commands cannot silently overwrite accepted state.
- Domain replay is deterministic and free from external side effects.
- Historical schemas remain readable after application upgrades.
- Events include suitable metadata for tracing and diagnostics.
- Projection updates are atomic or have a reliable asynchronous recovery mechanism.
- Projection lag and processing failures are monitored.
- Command retries use persistent idempotency identifiers where required.
- Snapshot strategy is based on measured replay performance.
- Projection rebuilds can be performed safely.
- Backups and restoration preserve event ordering and completeness.
- Security controls restrict modifications to authoritative history.
- Personal-data retention and deletion obligations are considered during event schema design.
- Integration consumers tolerate duplicate and out-of-order delivery.
- Tests cover replay, concurrency, historical schemas, and failure recovery.
- Operational runbooks document projection repair and corrupted-stream investigation.
Frequently Asked Questions
Event Sourcing is commonly associated with CQRS, Kafka, audit logging, and microservices. These are related concepts, but none should be treated as automatically interchangeable.
Does Event Sourcing Require CQRS?
No. Event Sourcing determines how authoritative state is persisted, while CQRS separates write and read responsibilities.
They often work well together because event histories can generate specialized read projections, but either pattern can be applied independently.
Is Kafka an Event Store?
Kafka provides durable event-log capabilities, but using Kafka does not automatically create a complete event-sourced persistence system.
An authoritative event store may require aggregate-level optimistic concurrency, long-term retention, precise version handling, event schema compatibility, and reliable reconstruction guarantees.
Kafka can participate in an Event Sourcing architecture when those requirements are intentionally designed and satisfied.
Is Event Sourcing the Same as an Audit Log?
No. An audit log records activity or changes, usually in addition to the application's authoritative current state.
Event Sourcing makes the event history itself authoritative and derives aggregate state from it.
Both can provide historical visibility, but their roles in application correctness are different.
Can Events Be Updated or Deleted?
Event Sourcing generally treats accepted events as immutable facts. Corrections are often represented by new compensating or corrective events instead of rewriting history.
However, real systems must also comply with applicable privacy, retention, and legal requirements. Event schema design should minimize sensitive information and provide a deliberate lifecycle strategy rather than assuming data can always be retained indefinitely.
Are Snapshots Required?
No. Snapshots are a performance optimization that becomes valuable when reconstructing aggregates from complete event histories is too expensive.
Small event streams can be replayed directly. Introduce snapshots when measurements justify the added storage and lifecycle complexity.
Is Event Sourcing Only for Microservices?
No. Event Sourcing can be implemented inside a monolith, modular monolith, microservice, or other application structure.
Its defining characteristic is authoritative event-based persistence, not the number of deployed services.
Conclusion
Event Sourcing changes how an application persists its business state. Instead of storing only the latest values, it records an ordered sequence of accepted domain events and reconstructs state by applying those events.
This makes historical reconstruction, auditability, and specialized projections possible, but introduces significant responsibilities around event compatibility, concurrency, replay performance, and operational recovery.
Event Sourcing can be combined with CQRS and event-driven architecture, but it does not require separate databases, a message broker, or microservices.
Choose it when historical business transitions are important enough to justify the additional complexity. For simple systems that only need the latest state, conventional persistence plus suitable audit records may be a better engineering decision.
Key Takeaway: Treat the event stream as authoritative business history, enforce aggregate-level ordering and concurrency, make replay deterministic, and maintain read projections as derived state that can be verified or reconstructed.
More Articles to Read
Continue exploring related SysDesPro articles:
- What Is the Unit of Work Pattern? Transactions and Consistency Explained — Coordinate multiple database changes within a single transaction.
- What Is Domain-Driven Design (DDD)? Concepts, Patterns, and Practical Examples — Understand aggregates, domain events, and business consistency boundaries.
- Transactional Outbox Pattern for Reliable Messaging — Keep committed business changes and outgoing events consistent.
- Event-Driven Architecture in Distributed Systems — Explore event producers, consumers, schemas, ordering, and failure recovery.
For further technical background, see Martin Fowler's Event Sourcing article, Martin Fowler's CQRS overview, and AWS Prescriptive Guidance for CQRS.
Comments (0)