当eventsourcing库记录聚合类路径时，如何迁移领域代码？

核心问题 #

事件存储中保存的事件topic是旧的全限定类名（比如event_sourcing_test.domain:Dog.Registered），聚合类迁移后，应用无法通过旧topic找到新位置的类，从而触发TopicError。

无需修改事件存储的解决方案 #

1. 注册单个Topic映射 #

使用库提供的register_topic_mapping函数，在应用启动阶段将旧topic直接映射到新的聚合/事件类，适合少量迁移场景。

示例代码：

from eventsourcing.utils import register_topic_mapping
from my_domain import Dog

# 映射旧的聚合类Topic
register_topic_mapping(
    "event_sourcing_test.domain:Dog",
    Dog
)

# 映射旧的事件类Topic
register_topic_mapping(
    "event_sourcing_test.domain:Dog.Registered",
    Dog.Registered
)

注意：这段代码必须在初始化repository或调用app.repository.get(...)之前执行，确保事件溯源时能解析旧Topic。

2. 自定义全局Topic解析器 #

如果有大量类需要迁移，可以自定义TopicResolver替换默认逻辑，批量处理旧Topic映射。

示例代码：

from eventsourcing.utils import TopicResolver, resolve_topic
from my_domain import Dog, Cat  # 假设还有其他迁移的类

class MigrationTopicResolver(TopicResolver):
    def resolve(self, topic: str) -> type:
        # 定义所有旧Topic到新类的映射
        migration_mappings = {
            "event_sourcing_test.domain:Dog": Dog,
            "event_sourcing_test.domain:Dog.Registered": Dog.Registered,
            "event_sourcing_test.domain:Cat": Cat,
            "event_sourcing_test.domain:Cat.Adopted": Cat.Adopted,
            # 添加更多迁移映射...
        }
        # 先检查是否是旧Topic，是则返回新类
        if topic in migration_mappings:
            return migration_mappings[topic]
        # 非迁移Topic用默认逻辑解析
        return super().resolve(topic)

# 替换全局的Topic解析器
resolve_topic.resolver = MigrationTopicResolver()

3. 聚合类重命名的处理 #

如果只是类名变更（比如Dog改为Canine），同样用register_topic_mapping完成映射：

from eventsourcing.utils import register_topic_mapping
from my_domain import Canine

register_topic_mapping(
    "event_sourcing_test.domain:Dog",
    Canine
)
register_topic_mapping(
    "event_sourcing_test.domain:Dog.Registered",
    Canine.Registered
)

原理说明 #

这些方法都是在运行时动态解析Topic，不需要修改事件存储中的旧数据。当库尝试解析旧Topic时，会通过映射关系找到新的类，从而正常重建聚合实例。

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