Prerequisites
- Basic understanding of programming concepts 📝
- Python installation (3.8+) 🐍
- VS Code or preferred IDE 💻
What you'll learn
- Understand the concept fundamentals 🎯
- Apply the concept in real projects 🏗️
- Debug common issues 🐛
- Write clean, Pythonic code ✨
🎯 Introduction
Welcome to this exciting tutorial on Event Sourcing! 🎉 In this guide, we’ll explore how to implement event sourcing patterns in Python - a powerful architectural pattern that stores all changes to application state as a sequence of events.
You’ll discover how event sourcing can transform your Python applications by providing complete audit trails, time-travel debugging, and the ability to reconstruct any past state. Whether you’re building financial systems 💰, e-commerce platforms 🛒, or distributed applications 🌐, understanding event sourcing is essential for writing scalable, maintainable systems.
By the end of this tutorial, you’ll feel confident implementing event sourcing in your own projects! Let’s dive in! 🏊♂️
📚 Understanding Event Sourcing
🤔 What is Event Sourcing?
Event sourcing is like keeping a detailed diary of everything that happens in your application 📖. Think of it as recording every single action (event) that changes your application’s state, rather than just storing the final result.
In Python terms, instead of updating a database record directly, you store events that describe what happened. This means you can:
- ✨ Replay events to reconstruct any past state
- 🚀 Debug issues by examining the exact sequence of events
- 🛡️ Never lose data - everything is preserved in the event log
💡 Why Use Event Sourcing?
Here’s why developers love event sourcing:
- Complete Audit Trail 🔒: Every change is recorded with who, what, when
- Time Travel 💻: Reconstruct application state at any point in time
- Event-Driven Architecture 📖: Natural fit for microservices and CQRS
- Business Intelligence 🔧: Rich data for analytics and reporting
Real-world example: Imagine building a banking system 🏦. With event sourcing, you can track every transaction, reconstruct account balances at any date, and provide complete audit trails for compliance.
🔧 Basic Syntax and Usage
📝 Simple Event Store Implementation
Let’s start with a friendly example:
# 👋 Hello, Event Sourcing!
from datetime import datetime
from typing import List, Dict, Any
from dataclasses import dataclass, field
import json
# 🎨 Creating a simple event
@dataclass
class Event:
event_type: str # 🎯 Type of event
aggregate_id: str # 🔑 Entity this event belongs to
data: Dict[str, Any] # 📦 Event payload
timestamp: datetime = field(default_factory=datetime.now) # ⏰ When it happened
version: int = 1 # 📊 Event version
# 🏗️ Basic Event Store
class EventStore:
def __init__(self):
self.events: List[Event] = [] # 📚 Our event log
def append(self, event: Event):
"""➕ Add event to the store"""
self.events.append(event)
print(f"📝 Stored event: {event.event_type} for {event.aggregate_id}")
def get_events(self, aggregate_id: str) -> List[Event]:
"""🔍 Get all events for an aggregate"""
return [e for e in self.events if e.aggregate_id == aggregate_id]💡 Explanation: Notice how we store events as immutable records! Each event captures what happened, when, and to which entity.
🎯 Common Event Patterns
Here are patterns you’ll use daily:
# 🏗️ Pattern 1: Domain Events
class AccountCreated(Event):
def __init__(self, account_id: str, owner: str, initial_balance: float):
super().__init__(
event_type="AccountCreated",
aggregate_id=account_id,
data={"owner": owner, "balance": initial_balance}
)
class MoneyDeposited(Event):
def __init__(self, account_id: str, amount: float):
super().__init__(
event_type="MoneyDeposited",
aggregate_id=account_id,
data={"amount": amount}
)
# 🎨 Pattern 2: Event Handlers
class EventHandler:
def handle(self, event: Event):
"""🔄 Process events based on type"""
handlers = {
"AccountCreated": self.handle_account_created,
"MoneyDeposited": self.handle_money_deposited
}
handler = handlers.get(event.event_type)
if handler:
handler(event)
def handle_account_created(self, event: Event):
print(f"✨ Creating account for {event.data['owner']}")
def handle_money_deposited(self, event: Event):
print(f"💰 Depositing ${event.data['amount']}")💡 Practical Examples
🛒 Example 1: Shopping Cart Event Store
Let’s build something real:
# 🛍️ Shopping cart with event sourcing
class ShoppingCart:
def __init__(self, cart_id: str, event_store: EventStore):
self.cart_id = cart_id
self.event_store = event_store
self.items = {} # 🛒 Current state
self._rebuild_state()
def _rebuild_state(self):
"""🔄 Replay events to rebuild current state"""
events = self.event_store.get_events(self.cart_id)
for event in events:
self._apply_event(event)
def _apply_event(self, event: Event):
"""🎯 Apply event to update state"""
if event.event_type == "ItemAdded":
item = event.data
if item['product_id'] in self.items:
self.items[item['product_id']]['quantity'] += item['quantity']
else:
self.items[item['product_id']] = item
elif event.event_type == "ItemRemoved":
product_id = event.data['product_id']
if product_id in self.items:
del self.items[product_id]
def add_item(self, product_id: str, name: str, price: float, quantity: int = 1):
"""➕ Add item to cart"""
event = Event(
event_type="ItemAdded",
aggregate_id=self.cart_id,
data={
"product_id": product_id,
"name": name,
"price": price,
"quantity": quantity,
"emoji": "🛍️"
}
)
self.event_store.append(event)
self._apply_event(event)
print(f"Added {event.data['emoji']} {name} to cart!")
def remove_item(self, product_id: str):
"""➖ Remove item from cart"""
event = Event(
event_type="ItemRemoved",
aggregate_id=self.cart_id,
data={"product_id": product_id}
)
self.event_store.append(event)
self._apply_event(event)
def get_total(self) -> float:
"""💰 Calculate total"""
return sum(item['price'] * item['quantity']
for item in self.items.values())
# 🎮 Let's use it!
store = EventStore()
cart = ShoppingCart("cart-123", store)
cart.add_item("1", "Python Book", 29.99)
cart.add_item("2", "Coffee", 4.99, quantity=3)
print(f"Total: ${cart.get_total():.2f}")🎯 Try it yourself: Add a checkout event and implement discount codes!
🎮 Example 2: Game State Event Sourcing
Let’s make it fun:
# 🏆 Game with event-sourced state
class GameEventStore:
def __init__(self):
self.events = []
self.snapshots = {} # 📸 State snapshots for performance
def append(self, event: Event):
self.events.append(event)
def create_snapshot(self, game_id: str, state: Dict):
"""📸 Save state snapshot"""
self.snapshots[game_id] = {
"state": state.copy(),
"version": len([e for e in self.events if e.aggregate_id == game_id])
}
class RPGGame:
def __init__(self, game_id: str, player_name: str, event_store: GameEventStore):
self.game_id = game_id
self.event_store = event_store
self.state = {
"player": player_name,
"level": 1,
"hp": 100,
"xp": 0,
"inventory": [],
"achievements": ["🌟 New Adventure"]
}
# 🎮 Start game event
start_event = Event(
event_type="GameStarted",
aggregate_id=game_id,
data={"player": player_name}
)
self.event_store.append(start_event)
def attack_monster(self, monster: str, damage: int):
"""⚔️ Battle event"""
event = Event(
event_type="MonsterDefeated",
aggregate_id=self.game_id,
data={
"monster": monster,
"damage": damage,
"xp_gained": damage * 10
}
)
self.event_store.append(event)
# 🎯 Update state
self.state["xp"] += damage * 10
print(f"⚔️ Defeated {monster}! Gained {damage * 10} XP!")
# 🎊 Check for level up
if self.state["xp"] >= self.state["level"] * 100:
self.level_up()
def level_up(self):
"""📈 Level up event"""
self.state["level"] += 1
self.state["hp"] = 100 + (self.state["level"] * 20)
event = Event(
event_type="LevelUp",
aggregate_id=self.game_id,
data={
"new_level": self.state["level"],
"new_hp": self.state["hp"]
}
)
self.event_store.append(event)
self.state["achievements"].append(f"🏆 Level {self.state['level']} Hero")
print(f"🎉 Level up! Now level {self.state['level']}!")
def replay_game(self) -> List[str]:
"""🔄 Replay all game events"""
events = [e for e in self.event_store.events if e.aggregate_id == self.game_id]
history = []
for event in events:
if event.event_type == "GameStarted":
history.append(f"🎮 {event.data['player']} started their adventure!")
elif event.event_type == "MonsterDefeated":
history.append(f"⚔️ Defeated {event.data['monster']}!")
elif event.event_type == "LevelUp":
history.append(f"🎉 Reached level {event.data['new_level']}!")
return history🚀 Advanced Concepts
🧙♂️ Advanced Topic 1: Event Sourcing with Projections
When you’re ready to level up, try this advanced pattern:
# 🎯 Advanced projection system
class Projection:
"""🔮 Build read models from events"""
def __init__(self):
self.handlers = {}
def register(self, event_type: str, handler):
"""📝 Register event handler"""
self.handlers[event_type] = handler
def project(self, events: List[Event]) -> Dict:
"""🏗️ Build projection from events"""
state = {}
for event in events:
handler = self.handlers.get(event.event_type)
if handler:
state = handler(state, event)
return state
# 🪄 Account balance projection
def create_account_projection():
projection = Projection()
def handle_account_created(state: Dict, event: Event) -> Dict:
state[event.aggregate_id] = {
"balance": event.data["balance"],
"owner": event.data["owner"],
"transactions": []
}
return state
def handle_money_deposited(state: Dict, event: Event) -> Dict:
if event.aggregate_id in state:
state[event.aggregate_id]["balance"] += event.data["amount"]
state[event.aggregate_id]["transactions"].append({
"type": "deposit",
"amount": event.data["amount"],
"timestamp": event.timestamp
})
return state
projection.register("AccountCreated", handle_account_created)
projection.register("MoneyDeposited", handle_money_deposited)
return projection🏗️ Advanced Topic 2: Event Store with Persistence
For the brave developers:
# 🚀 Persistent event store with JSON
import json
from pathlib import Path
class PersistentEventStore:
def __init__(self, filepath: str):
self.filepath = Path(filepath)
self.events = self._load_events()
def _load_events(self) -> List[Event]:
"""📂 Load events from file"""
if not self.filepath.exists():
return []
with open(self.filepath, 'r') as f:
data = json.load(f)
return [self._deserialize_event(e) for e in data]
def _deserialize_event(self, data: Dict) -> Event:
"""🔄 Convert JSON to Event"""
return Event(
event_type=data["event_type"],
aggregate_id=data["aggregate_id"],
data=data["data"],
timestamp=datetime.fromisoformat(data["timestamp"]),
version=data["version"]
)
def append(self, event: Event):
"""💾 Append and persist event"""
self.events.append(event)
self._save_events()
def _save_events(self):
"""💾 Save events to file"""
data = [{
"event_type": e.event_type,
"aggregate_id": e.aggregate_id,
"data": e.data,
"timestamp": e.timestamp.isoformat(),
"version": e.version
} for e in self.events]
with open(self.filepath, 'w') as f:
json.dump(data, f, indent=2)⚠️ Common Pitfalls and Solutions
😱 Pitfall 1: Mutable Events
# ❌ Wrong way - mutable events are dangerous!
event = Event("UserUpdated", "user-123", {"name": "Alice"})
event.data["name"] = "Bob" # 💥 Modifying event data!
# ✅ Correct way - events are immutable!
from dataclasses import dataclass, field
from typing import Dict, Any
@dataclass(frozen=True) # 🛡️ Make it immutable!
class ImmutableEvent:
event_type: str
aggregate_id: str
data: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
# 🔒 Make data immutable too
object.__setattr__(self, 'data', dict(self.data))🤯 Pitfall 2: Loading All Events
# ❌ Dangerous - loading millions of events!
def get_current_state(aggregate_id: str):
all_events = event_store.get_all_events() # 💥 Memory explosion!
return rebuild_from_events(all_events)
# ✅ Safe - use snapshots and pagination!
def get_current_state(aggregate_id: str):
# 📸 Start from latest snapshot
snapshot = event_store.get_latest_snapshot(aggregate_id)
# 🔄 Only replay events after snapshot
events_after_snapshot = event_store.get_events_after(
aggregate_id,
snapshot.version if snapshot else 0
)
state = snapshot.state if snapshot else {}
return apply_events(state, events_after_snapshot)🛠️ Best Practices
- 🎯 Events are Immutable: Never modify events after creation
- 📝 Event Names are Past Tense:
OrderPlacednotPlaceOrder - 🛡️ Version Your Events: Handle schema evolution gracefully
- 🎨 Keep Events Small: Store only what changed, not entire state
- ✨ Use Projections: Build read models for queries
🧪 Hands-On Exercise
🎯 Challenge: Build a Bank Account System
Create an event-sourced banking system:
📋 Requirements:
- ✅ Create accounts with initial balance
- 🏷️ Deposit and withdraw money
- 👤 Transfer between accounts
- 📅 Generate account statements
- 🎨 Calculate interest over time
🚀 Bonus Points:
- Add transaction limits
- Implement overdraft protection
- Create monthly statements from events
💡 Solution
🔍 Click to see solution
# 🎯 Event-sourced banking system!
from datetime import datetime, timedelta
from typing import List, Optional
class BankEventStore:
def __init__(self):
self.events = []
def append(self, event: Event):
self.events.append(event)
def get_account_events(self, account_id: str) -> List[Event]:
return [e for e in self.events if e.aggregate_id == account_id]
class BankAccount:
def __init__(self, account_id: str, owner: str, event_store: BankEventStore):
self.account_id = account_id
self.owner = owner
self.event_store = event_store
self.balance = 0.0
self.is_active = True
# 🎉 Create account event
create_event = Event(
event_type="AccountCreated",
aggregate_id=account_id,
data={
"owner": owner,
"initial_balance": 0.0,
"created_at": datetime.now().isoformat()
}
)
self.event_store.append(create_event)
def deposit(self, amount: float, description: str = ""):
"""💰 Deposit money"""
if amount <= 0:
raise ValueError("Amount must be positive")
event = Event(
event_type="MoneyDeposited",
aggregate_id=self.account_id,
data={
"amount": amount,
"description": description,
"balance_after": self.balance + amount
}
)
self.event_store.append(event)
self.balance += amount
print(f"💰 Deposited ${amount:.2f}. New balance: ${self.balance:.2f}")
def withdraw(self, amount: float, description: str = ""):
"""💸 Withdraw money"""
if amount <= 0:
raise ValueError("Amount must be positive")
if amount > self.balance:
raise ValueError("Insufficient funds!")
event = Event(
event_type="MoneyWithdrawn",
aggregate_id=self.account_id,
data={
"amount": amount,
"description": description,
"balance_after": self.balance - amount
}
)
self.event_store.append(event)
self.balance -= amount
print(f"💸 Withdrew ${amount:.2f}. New balance: ${self.balance:.2f}")
def transfer_to(self, target_account: 'BankAccount', amount: float):
"""💱 Transfer money to another account"""
# 📤 Withdrawal event
self.withdraw(amount, f"Transfer to {target_account.account_id}")
# 📥 Deposit event
target_account.deposit(amount, f"Transfer from {self.account_id}")
# 🔄 Transfer event for audit
transfer_event = Event(
event_type="MoneyTransferred",
aggregate_id=self.account_id,
data={
"from_account": self.account_id,
"to_account": target_account.account_id,
"amount": amount
}
)
self.event_store.append(transfer_event)
def get_statement(self, start_date: Optional[datetime] = None) -> List[str]:
"""📊 Generate account statement"""
events = self.event_store.get_account_events(self.account_id)
statement = [f"📊 Statement for {self.owner} (Account: {self.account_id})"]
statement.append("=" * 50)
running_balance = 0.0
for event in events:
if start_date and event.timestamp < start_date:
continue
if event.event_type == "AccountCreated":
statement.append(f"✨ Account opened")
elif event.event_type == "MoneyDeposited":
running_balance = event.data["balance_after"]
statement.append(
f"💰 {event.timestamp.strftime('%Y-%m-%d %H:%M')} | "
f"Deposit: ${event.data['amount']:.2f} | "
f"Balance: ${running_balance:.2f}"
)
elif event.event_type == "MoneyWithdrawn":
running_balance = event.data["balance_after"]
statement.append(
f"💸 {event.timestamp.strftime('%Y-%m-%d %H:%M')} | "
f"Withdrawal: ${event.data['amount']:.2f} | "
f"Balance: ${running_balance:.2f}"
)
statement.append("=" * 50)
statement.append(f"💰 Current Balance: ${self.balance:.2f}")
return statement
# 🎮 Test it out!
event_store = BankEventStore()
# Create accounts
alice_account = BankAccount("ACC-001", "Alice", event_store)
bob_account = BankAccount("ACC-002", "Bob", event_store)
# Transactions
alice_account.deposit(1000, "Initial deposit")
alice_account.withdraw(50, "ATM withdrawal")
alice_account.transfer_to(bob_account, 200)
# Generate statements
print("\n".join(alice_account.get_statement()))
print("\n")
print("\n".join(bob_account.get_statement()))🎓 Key Takeaways
You’ve learned so much! Here’s what you can now do:
- ✅ Implement event sourcing with confidence 💪
- ✅ Build event stores that preserve history 🛡️
- ✅ Create projections for read models 🎯
- ✅ Handle common pitfalls like a pro 🐛
- ✅ Design event-driven systems with Python! 🚀
Remember: Event sourcing is powerful but not always necessary. Use it when you need audit trails, temporal queries, or event-driven architectures! 🤝
🤝 Next Steps
Congratulations! 🎉 You’ve mastered event sourcing patterns!
Here’s what to do next:
- 💻 Practice with the banking exercise above
- 🏗️ Build an event-sourced order management system
- 📚 Learn about CQRS (Command Query Responsibility Segregation)
- 🌟 Explore event streaming with Kafka or RabbitMQ
Remember: Every distributed systems expert started with simple event stores. Keep coding, keep learning, and most importantly, have fun! 🚀
Happy coding! 🎉🚀✨