Python脚本事件溯源模式如何实现

本文目录导读：

1. 基础架构设计
2. 聚合根实现
3. 事件存储实现
4. 仓储模式实现
5. 投影和查询实现
6. 使用示例
7. 高级特性实现

我来详细解释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)

这个实现包含了事件溯源的核心概念,可以根据具体需求进行扩展和优化。