本文目录导读：
我来详细解释Python中事件溯源模式的实现。
基础架构设计
事件基类
from abc import ABC, abstractmethod
from datetime import datetime
from typing import Dict, Any, List, Optional
import uuid
import json
class Event(ABC):
"""事件基类"""
def __init__(self, aggregate_id: str):
self.event_id = str(uuid.uuid4())
self.aggregate_id = aggregate_id
self.timestamp = datetime.utcnow()
self.version = 0
@abstractmethod
def to_dict(self) -> Dict[str, Any]:
pass
@classmethod
@abstractmethod
def from_dict(cls, data: Dict[str, Any]):
pass
class AccountCreated(Event):
def __init__(self, aggregate_id: str, owner: str, initial_balance: float):
super().__init__(aggregate_id)
self.owner = owner
self.initial_balance = initial_balance
def to_dict(self):
return {
'event_type': 'AccountCreated',
'event_id': self.event_id,
'aggregate_id': self.aggregate_id,
'timestamp': self.timestamp.isoformat(),
'version': self.version,
'data': {
'owner': self.owner,
'initial_balance': self.initial_balance
}
}
@classmethod
def from_dict(cls, data):
event = cls(
data['aggregate_id'],
data['data']['owner'],
data['data']['initial_balance']
)
event.event_id = data['event_id']
event.timestamp = datetime.fromisoformat(data['timestamp'])
event.version = data['version']
return event
class MoneyDeposited(Event):
def __init__(self, aggregate_id: str, amount: float):
super().__init__(aggregate_id)
self.amount = amount
def to_dict(self):
return {
'event_type': 'MoneyDeposited',
'event_id': self.event_id,
'aggregate_id': self.aggregate_id,
'timestamp': self.timestamp.isoformat(),
'version': self.version,
'data': {
'amount': self.amount
}
}
@classmethod
def from_dict(cls, data):
event = cls(data['aggregate_id'], data['data']['amount'])
event.event_id = data['event_id']
event.timestamp = datetime.fromisoformat(data['timestamp'])
event.version = data['version']
return event
class MoneyWithdrawn(Event):
def __init__(self, aggregate_id: str, amount: float):
super().__init__(aggregate_id)
self.amount = amount
def to_dict(self):
return {
'event_type': 'MoneyWithdrawn',
'event_id': self.event_id,
'aggregate_id': self.aggregate_id,
'timestamp': self.timestamp.isoformat(),
'version': self.version,
'data': {
'amount': self.amount
}
}
@classmethod
def from_dict(cls, data):
event = cls(data['aggregate_id'], data['data']['amount'])
event.event_id = data['event_id']
event.timestamp = datetime.fromisoformat(data['timestamp'])
event.version = data['version']
return event
聚合根实现
class BankAccount:
"""银行账户聚合根"""
def __init__(self, aggregate_id: str = None):
self.aggregate_id = aggregate_id or str(uuid.uuid4())
self.owner = ""
self.balance = 0.0
self.version = 0
self.changes: List[Event] = []
def apply_event(self, event: Event):
"""应用事件到聚合根状态"""
event.version = self.version + 1
if isinstance(event, AccountCreated):
self._apply_account_created(event)
elif isinstance(event, MoneyDeposited):
self._apply_money_deposited(event)
elif isinstance(event, MoneyWithdrawn):
self._apply_money_withdrawn(event)
self.version = event.version
def _apply_account_created(self, event: AccountCreated):
self.owner = event.owner
self.balance = event.initial_balance
def _apply_money_deposited(self, event: MoneyDeposited):
self.balance += event.amount
def _apply_money_withdrawn(self, event: MoneyWithdrawn):
self.balance -= event.amount
# 业务方法
def create_account(self, owner: str, initial_balance: float):
if initial_balance < 0:
raise ValueError("Initial balance cannot be negative")
event = AccountCreated(self.aggregate_id, owner, initial_balance)
self.apply_event(event)
self.changes.append(event)
def deposit(self, amount: float):
if amount <= 0:
raise ValueError("Deposit amount must be positive")
event = MoneyDeposited(self.aggregate_id, amount)
self.apply_event(event)
self.changes.append(event)
def withdraw(self, amount: float):
if amount <= 0:
raise ValueError("Withdrawal amount must be positive")
if amount > self.balance:
raise ValueError("Insufficient funds")
event = MoneyWithdrawn(self.aggregate_id, amount)
self.apply_event(event)
self.changes.append(event)
def get_uncommitted_changes(self) -> List[Event]:
return self.changes
def mark_changes_as_committed(self):
self.changes = []
@classmethod
def from_events(cls, events: List[Event]):
"""从事件历史重建聚合根状态"""
account = cls()
for event in events:
account.apply_event(event)
account.mark_changes_as_committed()
return account
事件存储实现
from typing import Dict, List, Optional
import json
from datetime import datetime
class EventStore:
"""事件存储接口"""
def save_events(self, aggregate_id: str, events: List[Event], expected_version: int):
"""保存事件，带乐观锁控制"""
pass
def get_events(self, aggregate_id: str) -> List[Event]:
"""获取聚合的所有事件"""
pass
def get_all_events(self) -> List[Event]:
"""获取所有事件"""
pass
class InMemoryEventStore(EventStore):
"""内存事件存储实现"""
def __init__(self):
self.events: Dict[str, List[Dict]] = {}
self.event_registry = {
'AccountCreated': AccountCreated,
'MoneyDeposited': MoneyDeposited,
'MoneyWithdrawn': MoneyWithdrawn
}
def save_events(self, aggregate_id: str, events: List[Event], expected_version: int):
if aggregate_id not in self.events:
self.events[aggregate_id] = []
# 验证版本号
current_version = len(self.events[aggregate_id])
if current_version != expected_version:
raise ConcurrencyException(
f"Version conflict: expected {expected_version}, current {current_version}"
)
# 保存事件
for event in events:
self.events[aggregate_id].append(event.to_dict())
def get_events(self, aggregate_id: str) -> List[Event]:
if aggregate_id not in self.events:
return []
events = []
for event_data in self.events[aggregate_id]:
event_type = event_data['event_type']
event_class = self.event_registry[event_type]
events.append(event_class.from_dict(event_data))
return events
def get_all_events(self) -> List[Event]:
all_events = []
for aggregate_events in self.events.values():
for event_data in aggregate_events:
event_type = event_data['event_type']
event_class = self.event_registry[event_type]
all_events.append(event_class.from_dict(event_data))
# 按时间戳排序
all_events.sort(key=lambda e: e.timestamp)
return all_events
class ConcurrencyException(Exception):
"""并发异常"""
pass
仓储模式实现
class AccountRepository:
"""账户仓储"""
def __init__(self, event_store: EventStore):
self.event_store = event_store
def save(self, account: BankAccount):
"""保存账户的未提交更改"""
changes = account.get_uncommitted_changes()
if not changes:
return
# 期望版本 = 当前聚合版本 - 未提交事件数量
expected_version = account.version - len(changes)
self.event_store.save_events(
account.aggregate_id,
changes,
expected_version
)
account.mark_changes_as_committed()
def get_by_id(self, aggregate_id: str) -> Optional[BankAccount]:
"""通过ID获取账户"""
events = self.event_store.get_events(aggregate_id)
if not events:
return None
return BankAccount.from_events(events)
投影和查询实现
class AccountProjection:
"""账户投影 - 用于查询优化"""
def __init__(self):
self.accounts: Dict[str, Dict] = {}
def project(self, event: Event):
"""处理事件更新投影"""
if isinstance(event, AccountCreated):
self.accounts[event.aggregate_id] = {
'id': event.aggregate_id,
'owner': event.owner,
'balance': event.initial_balance,
'created_at': event.timestamp
}
elif isinstance(event, MoneyDeposited):
if event.aggregate_id in self.accounts:
self.accounts[event.aggregate_id]['balance'] += event.amount
elif isinstance(event, MoneyWithdrawn):
if event.aggregate_id in self.accounts:
self.accounts[event.aggregate_id]['balance'] -= event.amount
def get_account(self, aggregate_id: str) -> Optional[Dict]:
return self.accounts.get(aggregate_id)
def get_all_accounts(self) -> List[Dict]:
return list(self.accounts.values())
def rebuild_from_events(self, events: List[Event]):
"""从事件历史重建投影"""
self.accounts = {}
for event in events:
self.project(event)
class AccountQueryService:
"""账户查询服务"""
def __init__(self, projection: AccountProjection):
self.projection = projection
def get_balance(self, account_id: str) -> Optional[float]:
account = self.projection.get_account(account_id)
if account:
return account['balance']
return None
def get_account_summary(self, account_id: str) -> Optional[Dict]:
return self.projection.get_account(account_id)
使用示例
def main():
# 初始化组件
event_store = InMemoryEventStore()
repository = AccountRepository(event_store)
projection = AccountProjection()
query_service = AccountQueryService(projection)
# 1. 创建账户
print("=== 创建账户 ===")
account = BankAccount()
account.create_account("John Doe", 1000.0)
repository.save(account)
print(f"Created account {account.aggregate_id} with balance {account.balance}")
# 2. 存款
print("\n=== 存款操作 ===")
account.deposit(500.0)
repository.save(account)
print(f"Deposited 500. Balance: {account.balance}")
# 3. 取款
print("\n=== 取款操作 ===")
account.withdraw(200.0)
repository.save(account)
print(f"Withdrew 200. Balance: {account.balance}")
# 4. 更新投影
print("\n=== 更新投影 ===")
all_events = event_store.get_all_events()
projection.rebuild_from_events(all_events)
# 5. 查询操作
print("\n=== 查询操作 ===")
summary = query_service.get_account_summary(account.aggregate_id)
print(f"Account Summary: {summary}")
balance = query_service.get_balance(account.aggregate_id)
print(f"Current Balance: {balance}")
# 6. 从事件历史重建聚合
print("\n=== 从事件历史重建 ===")
loaded_account = repository.get_by_id(account.aggregate_id)
print(f"Loaded account: {loaded_account.aggregate_id}")
print(f"Owner: {loaded_account.owner}")
print(f"Balance: {loaded_account.balance}")
print(f"Version: {loaded_account.version}")
if __name__ == "__main__":
main()
高级特性实现
事件版本控制
class EventVersion:
def __init__(self):
self.version = 0
self.upgraders = {}
def register_upgrader(self, from_version: int, to_version: int, upgrader):
"""注册事件升级器"""
self.upgraders[(from_version, to_version)] = upgrader
def upgrade_events(self, events: List[Event]) -> List[Event]:
"""升级事件到最新版本"""
upgraded_events = []
for event in events:
current_version = event.version
while current_version < self.version:
next_version = current_version + 1
if (current_version, next_version) in self.upgraders:
event = self.upgraders[(current_version, next_version)](event)
current_version = next_version
upgraded_events.append(event)
return upgraded_events
快照管理
class Snapshot:
def __init__(self, aggregate_id: str, version: int, state: Dict):
self.aggregate_id = aggregate_id
self.version = version
self.state = state
self.created_at = datetime.utcnow()
class SnapshotRepository:
def __init__(self, snapshot_threshold: int = 100):
self.snapshots: Dict[str, Snapshot] = {}
self.threshold = snapshot_threshold
def save_snapshot(self, account: BankAccount):
"""保存快照"""
if account.version >= self.threshold and account.version % self.threshold == 0:
snapshot = Snapshot(
account.aggregate_id,
account.version,
{
'owner': account.owner,
'balance': account.balance
}
)
self.snapshots[account.aggregate_id] = snapshot
def get_snapshot(self, aggregate_id: str) -> Optional[Snapshot]:
"""获取快照"""
return self.snapshots.get(aggregate_id)
这个实现包含了事件溯源的核心概念,可以根据具体需求进行扩展和优化。
