Python Event Sourcing实践：如何实现Projection查询聚合列表？

问题描述 #

我正在学习Event Sourcing（事件溯源）设计模式，使用python eventsourcing库进行开发。已完成《Dog School》教程，但仍不清楚如何实现Projection（投影）。我认为若要查询聚合中的所有条目，需借助投影功能？似乎该库未内置投影，是否需自行编写查询语句从持久化层获取数据？

教程中的DogSchool类代码如下：

class DogSchool(Application):
    def register_dog(self, name):
        dog = Dog(name)
        self.save(dog)
        return dog.id

    def add_trick(self, dog_id, trick):
        dog = self.repository.get(dog_id)
        dog.add_trick(trick=trick)
        self.save(dog)

    def get_dog(self, dog_id):
        dog = self.repository.get(dog_id)
        return {'name': dog.name, 'tricks': tuple(dog.tricks)}

我希望实现如下的list_dogs方法，用于获取所有狗的列表：

def list_dogs(self):
        dogs = get_all_from_aggregate() #??
        return dogs

解决方案 #

eventsourcing库确实没有提供直接查询所有聚合的开箱即用API，但可以通过两种方式实现需求：

方式一：直接从仓库加载所有聚合（适合小型场景）

利用库的repository.get_all_ids()方法获取指定聚合类的所有ID，再逐个加载聚合并转换为所需格式：

class DogSchool(Application):
    # 保留原有的register_dog、add_trick、get_dog方法

    def list_dogs(self):
        # 获取所有Dog类型聚合的ID
        dog_ids = self.repository.get_all_ids(aggregate_class=Dog)
        dogs = []
        for dog_id in dog_ids:
            dog = self.repository.get(dog_id)
            dogs.append({
                'id': dog_id,
                'name': dog.name,
                'tricks': tuple(dog.tricks)
            })
        return dogs

这种方式实现简单，无需额外代码；但如果聚合数量多，逐个加载会影响性能——因为每个聚合都需要从事件流重建状态。

方式二：实现Projection投影（适合查询频繁/大数据场景）

投影的核心是维护一个只读视图，实时同步聚合的事件变化，查询时直接从视图获取数据，避免重复重建聚合。

步骤1：确保Dog聚合定义了对应事件

确认你的Dog聚合类已定义事件（教程中可能已包含，这里补充完整）：

from eventsourcing.domain import AggregateRoot, Event

class DogRegistered(Event):
    name: str

class TrickAdded(Event):
    trick: str

class Dog(AggregateRoot):
    def __init__(self, name):
        super().__init__()
        self.name = name
        self.tricks = []
        self.__trigger_event__(DogRegistered(name=name))

    def add_trick(self, trick):
        self.tricks.append(trick)
        self.__trigger_event__(TrickAdded(trick=trick))

步骤2：创建投影类

定义投影类监听Dog的事件，维护存储视图：

from eventsourcing.domain import Projection

class DogProjection(Projection):
    def __init__(self):
        # 用字典存储视图，key为狗的ID，value为狗的信息
        self.dog_view = {}

    # 处理DogRegistered事件，初始化视图条目
    def apply_DogRegistered(self, event):
        self.dog_view[event.originator_id] = {
            'id': event.originator_id,
            'name': event.name,
            'tricks': []
        }

    # 处理TrickAdded事件，更新视图中的技巧列表
    def apply_TrickAdded(self, event):
        if event.originator_id in self.dog_view:
            self.dog_view[event.originator_id]['tricks'].append(event.trick)

步骤3：在Application中注册投影

修改DogSchool类，初始化并注册投影，让事件保存时自动触发投影更新：

class DogSchool(Application):
    def __init__(self):
        super().__init__()
        # 初始化投影
        self.dog_projection = DogProjection()
        # 订阅投影，监听Dog的所有事件
        self.event_store.subscribe(
            self.dog_projection,
            predicate=lambda e: isinstance(e, (DogRegistered, TrickAdded))
        )

    # 保留原有的register_dog、add_trick、get_dog方法

    def list_dogs(self):
        # 直接从投影视图获取所有狗的数据
        return list(self.dog_projection.dog_view.values())

这种方式查询性能高，视图实时更新；如果需要持久化视图（比如存到数据库），可以把字典替换为SQLAlchemy模型或其他持久化存储，在apply方法中操作数据库即可。

内容的提问来源于stack exchange，提问作者d s