基于事件溯源的数据血缘追踪在可观测性体系中的实现路径

event-sourcing-lineage-demo/
├── config.py
├── run.py
├── requirements.txt
├── event_store.db
├── services/
│   ├── __init__.py
│   ├── event_store.py
│   ├── lineage_graph.py
│   └── event_bus.py
├── models/
│   ├── __init__.py
│   ├── events.py
│   └── entities.py
└── handlers/
    ├── __init__.py
    ├── user_service.py
    ├── order_service.py
    └── lineage_query.py

import json
from datetime import datetime
from dataclasses import dataclass, asdict, field
from typing import Any, Dict, List, Optional
from uuid import uuid4

@dataclass
class DomainEvent:
    """领域事件基类"""
    event_id: str = field(default_factory=lambda: str(uuid4()))
    event_type: str = field(init=False)
    aggregate_id: str  # 发生变更的聚合根ID， 如 "user_123"
    aggregate_type: str  # 聚合根类型， 如 "User"
    event_data: Dict[str, Any]  # 事件负载，包含变更详情
    timestamp: datetime = field(default_factory=datetime.utcnow)
    # 用于血缘分析的关键字段
    affected_entity_id: str  # 主要被影响的数据实体ID，通常等于aggregate_id
    source_entity_ids: List[str] = field(default_factory=list) # 导致此变更的源实体ID列表

    def __post_init__(self):
        self.event_type = self.__class__.__name__

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典，用于序列化"""
        data = asdict(self)
        data['timestamp'] = self.timestamp.isoformat()
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'DomainEvent':
        """从字典反序列化"""
        data['timestamp'] = datetime.fromisoformat(data['timestamp'])
        # 动态查找具体的DomainEvent子类
        event_class = globals().get(data['event_type'], DomainEvent)
        return event_class(**data)


# 具体领域事件定义
@dataclass
class UserProfileUpdated(DomainEvent):
    def __post_init__(self):
        super().__post_init__()
        self.aggregate_type = "User"
        # affected_entity_id 在创建时传入， 例如 "user_001"
        # source_entity_ids 可能为空，表示初始创建或外部触发


@dataclass
class OrderCreated(DomainEvent):
    def __post_init__(self):
        super().__post_init__()
        self.aggregate_type = "Order"
        # event_data 应包含 ‘user_id‘
        # source_entity_ids 应包含引用的用户ID


@dataclass
class OrderAmended(DomainEvent):
    """订单修订事件，可能由用户信息更新触发"""
    def __post_init__(self):
        super().__post_init__()
        self.aggregate_type = "Order"

import sqlite3
import json
from typing import List, Optional
from models.events import DomainEvent

class EventStore:
    """简单的事件存储实现"""
    
    def __init__(self, db_path: str = 'event_store.db'):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        """初始化数据库表"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS events (
                event_id TEXT PRIMARY KEY,
                event_type TEXT NOT NULL,
                aggregate_id TEXT NOT NULL,
                aggregate_type TEXT NOT NULL,
                event_data TEXT NOT NULL, -- JSON存储
                timestamp DATETIME NOT NULL,
                affected_entity_id TEXT NOT NULL,
                source_entity_ids TEXT NOT NULL -- JSON数组存储
            )
        ''')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_aggregate ON events (aggregate_type, aggregate_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_affected ON events (affected_entity_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_timestamp ON events (timestamp)')
        conn.commit()
        conn.close()

    def append(self, event: DomainEvent):
        """追加一个事件到存储"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        event_dict = event.to_dict()
        cursor.execute('''
            INSERT INTO events (event_id, event_type, aggregate_id, aggregate_type, 
                               event_data, timestamp, affected_entity_id, source_entity_ids)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            event_dict['event_id'],
            event_dict['event_type'],
            event_dict['aggregate_id'],
            event_dict['aggregate_type'],
            json.dumps(event_dict['event_data']),
            event_dict['timestamp'],
            event_dict['affected_entity_id'],
            json.dumps(event_dict['source_entity_ids'])
        ))
        conn.commit()
        conn.close()

    def get_events_for_aggregate(self, aggregate_type: str, aggregate_id: str) -> List[DomainEvent]:
        """获取指定聚合根的所有事件（用于重建状态）"""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        cursor.execute('''
            SELECT * FROM events 
            WHERE aggregate_type = ? AND aggregate_id = ?
            ORDER BY timestamp ASC
        ''', (aggregate_type, aggregate_id))
        rows = cursor.fetchall()
        conn.close()
        events = []
        for row in rows:
            event_dict = dict(row)
            event_dict['event_data'] = json.loads(event_dict['event_data'])
            event_dict['source_entity_ids'] = json.loads(event_dict['source_entity_ids'])
            events.append(DomainEvent.from_dict(event_dict))
        return events

    def get_all_events(self, limit: Optional[int] = None) -> List[DomainEvent]:
        """获取所有事件（用于血缘图重建）"""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        sql = 'SELECT * FROM events ORDER BY timestamp ASC'
        if limit:
            sql += f' LIMIT {limit}'
        cursor.execute(sql)
        rows = cursor.fetchall()
        conn.close()
        events = []
        for row in rows:
            event_dict = dict(row)
            event_dict['event_data'] = json.loads(event_dict['event_data'])
            event_dict['source_entity_ids'] = json.loads(event_dict['source_entity_ids'])
            events.append(DomainEvent.from_dict(event_dict))
        return events

    def get_events_by_affected_entity(self, entity_id: str) -> List[DomainEvent]:
        """获取直接影响指定实体的事件"""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        cursor.execute('''
            SELECT * FROM events 
            WHERE affected_entity_id = ? 
            ORDER BY timestamp ASC
        ''', (entity_id,))
        rows = cursor.fetchall()
        conn.close()
        # ... 反序列化逻辑同 get_all_events ...
        return [self._row_to_event(row) for row in rows]
    
    def _row_to_event(self, row) -> DomainEvent:
        """辅助方法：从数据库行反序列化为事件对象"""
        event_dict = dict(row)
        event_dict['event_data'] = json.loads(event_dict['event_data'])
        event_dict['source_entity_ids'] = json.loads(event_dict['source_entity_ids'])
        return DomainEvent.from_dict(event_dict)

from typing import Dict, List, Set, Tuple, Optional
from collections import defaultdict, deque
import networkx as nx # 用于图操作，需安装networkx库。为减少依赖，此处提供简化实现。
# 注意：为控制代码量，此处实现一个简化的图结构，不使用networkx。

class SimpleLineageGraph:
    """简化的血缘图实现（邻接表表示有向图）"""
    
    def __init__(self):
        # 正向邻接表： key -> [values] 表示 key 影响/生成了 values (下游)
        self._downstream: Dict[str, Set[str]] = defaultdict(set)
        # 反向邻接表： key -> [values] 表示 key 依赖于 values (上游)
        self._upstream: Dict[str, Set[str]] = defaultdict(set)
        # 存储边上的元数据（事件ID）
        self._edge_metadata: Dict[Tuple[str, str], List[str]] = defaultdict(list)

    def add_edge(self, source_entity_id: str, target_entity_id: str, event_id: str):
        """添加一条从源实体到目标实体的边"""
        if source_entity_id == target_entity_id:
            return
        self._downstream[source_entity_id].add(target_entity_id)
        self._upstream[target_entity_id].add(source_entity_id)
        self._edge_metadata[(source_entity_id, target_entity_id)].append(event_id)

    def get_upstream(self, entity_id: str, depth: Optional[int] = None) -> Dict[str, List[str]]:
        """获取实体所有上游依赖（来源），返回 {实体ID: 路径事件ID列表} 的字典"""
        visited = {entity_id: []}
        queue = deque([(entity_id, [])]) # (当前节点， 路径事件ID列表)
        
        while queue:
            current, path = queue.popleft()
            if depth is not None and len(path) >= depth:
                continue
            for source in self._upstream.get(current, set()):
                edge_key = (source, current)
                if edge_key in self._edge_metadata:
                    for evt_id in self._edge_metadata[edge_key]:
                        new_path = path + [evt_id]
                        if source not in visited: # 简单处理，不记录所有路径
                            visited[source] = new_path
                            queue.append((source, new_path))
        visited.pop(entity_id, None) # 移除自己
        return visited

    def get_downstream(self, entity_id: str, depth: Optional[int] = None) -> Dict[str, List[str]]:
        """获取实体所有下游影响"""
        visited = {entity_id: []}
        queue = deque([(entity_id, [])])
        
        while queue:
            current, path = queue.popleft()
            if depth is not None and len(path) >= depth:
                continue
            for target in self._downstream.get(current, set()):
                edge_key = (current, target)
                if edge_key in self._edge_metadata:
                    for evt_id in self._edge_metadata[edge_key]:
                        new_path = path + [evt_id]
                        if target not in visited:
                            visited[target] = new_path
                            queue.append((target, new_path))
        visited.pop(entity_id, None)
        return visited

    def rebuild_from_events(self, events: List['DomainEvent']):
        """从事件列表重建血缘图（用于初始化或回放）"""
        self._downstream.clear()
        self._upstream.clear()
        self._edge_metadata.clear()
        for event in events:
            for source_id in event.source_entity_ids:
                self.add_edge(source_id, event.affected_entity_id, event.event_id)

class LineageGraphManager:
    """血缘图管理服务"""
    
    def __init__(self, event_store: 'EventStore'):
        self.event_store = event_store
        self.graph = SimpleLineageGraph()
        self._rebuild_graph()

    def _rebuild_graph(self):
        """从事件存储中重建完整的血缘图"""
        all_events = self.event_store.get_all_events()
        self.graph.rebuild_from_events(all_events)
        print(f"[LineageGraph] Rebuilt graph with {len(all_events)} events.")

    def process_new_event(self, event: 'DomainEvent'):
        """处理一个新事件，更新血缘图"""
        for source_id in event.source_entity_ids:
            self.graph.add_edge(source_id, event.affected_entity_id, event.event_id)
        print(f"[LineageGraph] Processed event {event.event_type} for {event.affected_entity_id}")

    def query_lineage(self, entity_id: str, direction: str = 'both', depth: Optional[int] = None) -> Dict:
        """查询数据血缘"""
        result = {}
        if direction in ('upstream', 'both'):
            result['upstream'] = self.graph.get_upstream(entity_id, depth)
        if direction in ('downstream', 'both'):
            result['downstream'] = self.graph.get_downstream(entity_id, depth)
        return result

from flask import Blueprint, request, jsonify
from models.events import UserProfileUpdated
from services.event_store import EventStore
from services.event_bus import event_bus
import uuid

user_bp = Blueprint('user_service', __name__, url_prefix='/api/users')
# 为了简化，我们在内存中维护一个"当前状态"快照，实际应从事件重建。
user_snapshots = {} 

@user_bp.route('/<user_id>', methods=['PUT'])
def update_user(user_id):
    """更新用户信息，并发布领域事件"""
    data = request.get_json()
    if not data:
        return jsonify({'error': 'Invalid data'}), 400
    
    # 1. 模拟业务逻辑：更新快照
    user_snapshots[user_id] = {
        'user_id': user_id,
        'name': data.get('name', 'Unknown'),
        'email': data.get('email', ''),
        'version': user_snapshots.get(user_id, {}).get('version', 0) + 1
    }
    
    # 2. 构造并发布事件
    event = UserProfileUpdated(
        aggregate_id=user_id,
        event_data={'old_email': user_snapshots[user_id].get('old_email'), **data}, # 示例，实际应更严谨
        affected_entity_id=user_id,
        source_entity_ids=[]  # 用户更新可能没有明确的内部数据源
    )
    
    # 3. 持久化事件
    event_store = EventStore()
    event_store.append(event)
    
    # 4. 通知事件总线（触发订阅者处理并更新血缘图）
    event_bus.publish(event)
    
    return jsonify({
        'message': 'User updated',
        'user': user_snapshots[user_id],
        'event_id': event.event_id
    })

@user_bp.route('/<user_id>', methods=['GET'])
def get_user(user_id):
    """获取用户当前状态"""
    user = user_snapshots.get(user_id)
    if user:
        return jsonify(user)
    else:
        # 尝试从事件重建（简化演示，仅返回空）
        return jsonify({'error': 'User not found'}), 404

from flask import Blueprint, request, jsonify
from models.events import OrderCreated, OrderAmended, UserProfileUpdated
from services.event_store import EventStore
from services.event_bus import event_bus
import uuid

order_bp = Blueprint('order_service', __name__, url_prefix='/api/orders')
order_snapshots = {}

def _rebuild_order(order_id):
    """从事件流重建订单状态（简化演示，仅获取最后一个快照）"""
    # 实际应回放所有 Order* 事件
    return order_snapshots.get(order_id)

@order_bp.route('', methods=['POST'])
def create_order():
    """创建新订单"""
    data = request.get_json()
    user_id = data.get('user_id')
    if not user_id:
        return jsonify({'error': 'user_id is required'}), 400
    
    order_id = f"order_{uuid.uuid4().hex[:8]}"
    
    # 1. 创建订单快照
    order_snapshots[order_id] = {
        'order_id': order_id,
        'user_id': user_id,
        'items': data.get('items', []),
        'status': 'CREATED',
        'user_name': 'N/A', # 初始可能没有用户名
        'version': 1
    }
    
    # 2. 构造并发布 OrderCreated 事件
    # 注意：source_entity_ids 包含了本订单所依赖的用户ID
    event = OrderCreated(
        aggregate_id=order_id,
        event_data=order_snapshots[order_id],
        affected_entity_id=order_id,
        source_entity_ids=[user_id]  # 关键：订单依赖于用户
    )
    event_store = EventStore()
    event_store.append(event)
    event_bus.publish(event)
    
    return jsonify({
        'message': 'Order created',
        'order': order_snapshots[order_id],
        'event_id': event.event_id
    })

# --- 事件处理器 ---
@event_bus.subscribe(UserProfileUpdated)
def handle_user_updated(event: UserProfileUpdated):
    """订阅用户更新事件，更新相关订单，并产生新的修订事件"""
    updated_user_id = event.affected_entity_id
    event_store = EventStore()
    
    # 查找所有依赖于该用户的订单（在实际中，可能需要一个查询索引）
    # 这里简化：遍历所有订单快照（仅用于演示，生产环境效率低）
    for order_id, order in order_snapshots.items():
        if order['user_id'] == updated_user_id:
            # 1. 更新订单快照中的用户信息
            order['user_name'] = event.event_data.get('name', order['user_name'])
            order['version'] += 1
            # 2. 发布 OrderAmended 事件，记录这次由用户更新触发的修订
            amend_event = OrderAmended(
                aggregate_id=order_id,
                event_data={
                    'reason': 'USER_PROFILE_UPDATE',
                    'updated_fields': ['user_name'],
                    'triggered_by_event': event.event_id
                },
                affected_entity_id=order_id,
                source_entity_ids=[updated_user_id]  # 关键：修订来源于用户更新
            )
            event_store.append(amend_event)
            event_bus.publish(amend_event) # 发布修订事件，进一步更新血缘图
            print(f"[OrderService] Order {order_id} amended due to user {updated_user_id} update.")

from flask import Blueprint, request, jsonify
from services.lineage_graph import LineageGraphManager
from services.event_store import EventStore

lineage_bp = Blueprint('lineage_query', __name__, url_prefix='/api/lineage')

# 初始化管理器
_event_store = EventStore()
_lineage_manager = LineageGraphManager(_event_store)

@lineage_bp.route('/<entity_id>')
def get_lineage(entity_id):
    direction = request.args.get('direction', 'both')  # upstream, downstream, both
    depth = request.args.get('depth', type=int)  # 可选深度限制
    result = _lineage_manager.query_lineage(entity_id, direction, depth)
    return jsonify({
        'entity_id': entity_id,
        'lineage': result
    })

@lineage_bp.route('/events/<entity_id>')
def get_events_for_entity(entity_id):
    """获取直接影响该实体的事件列表"""
    event_store = EventStore()
    events = event_store.get_events_by_affected_entity(entity_id)
    return jsonify({
        'entity_id': entity_id,
        'events': [e.to_dict() for e in events]
    })

from typing import Dict, List, Callable, Type
from models.events import DomainEvent

class EventBus:
    """简单的事件总线（进程内）"""
    
    def __init__(self):
        self._subscribers: Dict[str, List[Callable]] = defaultdict(list)

    def subscribe(self, event_class: Type[DomainEvent]):
        """装饰器，用于订阅特定类型的事件"""
        def decorator(handler: Callable):
            self._subscribers[event_class.__name__].append(handler)
            return handler
        return decorator

    def publish(self, event: DomainEvent):
        """发布一个事件，通知所有订阅者"""
        handlers = self._subscribers.get(event.event_type, [])
        for handler in handlers:
            try:
                handler(event)
            except Exception as e:
                print(f"Error handling event {event.event_type} in {handler.__name__}: {e}")

# 全局事件总线实例
event_bus = EventBus()

from flask import Flask
import threading
import time
from services.event_store import EventStore
from services.lineage_graph import LineageGraphManager

# 导入蓝图
from handlers.user_service import user_bp
from handlers.order_service import order_bp
from handlers.lineage_query import lineage_bp

def create_app():
    app = Flask(__name__)
    
    # 注册蓝图
    app.register_blueprint(user_bp)
    app.register_blueprint(order_bp)
    app.register_blueprint(lineage_query.lineage_bp)
    
    @app.route('/')
    def index():
        return '<h1>Event Sourcing & Data Lineage Demo</h1><p>Check /api/users, /api/orders, /api/lineage</p>'
    
    return app

def start_lineage_manager():
    """在后台线程中启动并运行血缘图管理器（模拟实时处理）"""
    # 注意：实际生产环境中，管理器应作为独立服务或工作线程，从事件流（如Kafka）消费
    time.sleep(1) # 等待应用初始化
    event_store = EventStore()
    manager = LineageGraphManager(event_store)
    # 简化：管理器已初始化并重建图。实时事件通过 event_bus.publish 触发 process_new_event。
    # 这里我们保持线程运行，实际可能是一个事件循环。
    print("[Main] Lineage manager started.")
    while True:
        time.sleep(10) # 模拟后台运行
        # 可以定期执行一些维护任务

if __name__ == '__main__':
    # 启动后台血缘管理线程（简化演示）
    lineage_thread = threading.Thread(target=start_lineage_manager, daemon=True)
    lineage_thread.start()
    
    # 启动Flask应用
    app = create_app()
    app.run(host='0.0.0.0', port=5000, debug=True)

import os

BASE_DIR = os.path.abspath(os.path.dirname(__file__))

class Config:
    EVENT_STORE_DB_PATH = os.path.join(BASE_DIR, 'event_store.db')

Flask==2.3.3
networkx==3.1  # 可选，本示例未直接使用其高级API

mkdir event-sourcing-lineage-demo
    cd event-sourcing-lineage-demo

pip install -r requirements.txt

python run.py

    [LineageGraph] Rebuilt graph with 0 events.
    [Main] Lineage manager started.

     * Serving Flask app 'run'
     * Debug mode: on
     * Running on all addresses (0.0.0.0)
     * Running on http://127.0.0.1:5000

curl -X PUT http://127.0.0.1:5000/api/users/user_001 \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice", "email": "alice@example.com"}'

curl -X POST http://127.0.0.1:5000/api/orders \
  -H "Content-Type: application/json" \
  -d '{"user_id": "user_001", "items": ["item1", "item2"]}'

curl -X PUT http://127.0.0.1:5000/api/users/user_001 \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice Smith", "email": "alice.smith@example.com"}'

# 查询订单的上游来源（应看到用户 user_001）
curl http://127.0.0.1:5000/api/lineage/order_abc123?direction=upstream
# 查询用户的下游影响（应看到订单 order_abc123）
curl http://127.0.0.1:5000/api/lineage/user_001?direction=downstream

curl http://127.0.0.1:5000/api/lineage/events/order_abc123