使用Django进行事件溯源

此包支持使用Python eventsourcing 库与 Django ORM 一起使用。

要使用Django与您的Python事件溯源应用程序一起使用

• 安装Python包 eventsourcing_django
• 将 'eventsourcing_django' 添加到您的Django项目中的 INSTALLED_APPS 设置
• 迁移此Django应用的数据库
• 设置环境变量 PERSISTENCE_MODULE 为 'eventsourcing_django'

以下为更多信息。

安装

使用pip从Python包索引安装稳定版。请注意，建议将Python包安装到Python虚拟环境中。

$ pip install eventsourcing_django

Django

如果您正在使用Django 3.2或更高版本，请将 'eventsourcing_django' 添加到您的Django项目的 INSTALLED_APPS 设置中。

INSTALLED_APPS = [
    ...
    'eventsourcing_django',
]

如果您正在使用Django 2.2、3.0或3.1，请将 'eventsourcing_django.apps.EventsourcingConfig' 添加到您的Django项目的 INSTALLED_APPS 设置中。

INSTALLED_APPS = [
    ...
    'eventsourcing_django.apps.EventsourcingConfig',
]

要迁移数据库，请运行Django的 manage.py migrate 命令。

$ python manage.py migrate eventsourcing_django

事件溯源

以常规方式定义聚合和应用。

from eventsourcing.application import Application
from eventsourcing.domain import Aggregate, event
from uuid import uuid5, NAMESPACE_URL


class TrainingSchool(Application):
    def register(self, name):
        dog = Dog(name)
        self.save(dog)

    def add_trick(self, name, trick):
        dog = self.repository.get(Dog.create_id(name))
        dog.add_trick(trick)
        self.save(dog)

    def get_tricks(self, name):
        dog = self.repository.get(Dog.create_id(name))
        return dog.tricks


class Dog(Aggregate):
    @event('Registered')
    def __init__(self, name):
        self.name = name
        self.tricks = []

    @staticmethod
    def create_id(name):
        return uuid5(NAMESPACE_URL, f'/dogs/{name}')

    @event('TrickAdded')
    def add_trick(self, trick):
        self.tricks.append(trick)

按照常规方式构建和使用应用程序。在应用程序的环境中设置 PERSISTENCE_MODULE 为 'eventsourcing_django'。您可能希望在Django项目“就绪”时在信号上构建应用程序对象。您可以使用Django应用的 apps.py 模块中 AppConfig 类的 ready() 方法。

school = TrainingSchool(env={
    "PERSISTENCE_MODULE": "eventsourcing_django",
})

应用程序的方法可以从Django视图和表单中调用。

school.register('Fido')
school.add_trick('Fido', 'roll over')
school.add_trick('Fido', 'play dead')
tricks = school.get_tricks('Fido')
assert tricks == ['roll over', 'play dead']

有关更多信息，请参阅Python的 eventsourcing 库和 Django 项目。

管理命令

Django应用 eventsourcing_django 附带了以下管理命令。

同步跟随者

手动将跟随者（即 ProcessApplication 实例）与其所有领导者同步，这些领导者由 eventsourcing.system.System 的管道定义。

用法

$ python manage.py sync_followers [-n] [-v {0,1,2,3}] [follower [follower ...]]

其中 follower 表示要同步的跟随者名称。不指定任何内容表示同步系统中找到的所有 跟随者。

相关选项

• -n，--dry-run：加载并处理所选跟随者的所有未见事件，但最终回滚所有更改。
• -v {0,1,2,3}，--verbosity {0,1,2,3}：详细程度；0=最小输出，1=常规输出，2=详细输出，3=非常详细输出。

要获取完整选项列表，请将 --help 标志传递给命令。

示例

• 同步运行器中找到的所有跟随者

  $ python manage.py sync_followers
  

• 同步单个跟随者

  $ python manage.py sync_followers TrainingSchool
  

该命令支持常规的 -v/--verbosity 可选参数，以及 -n/--dry-run 标志。

注意，以干燥运行模式运行命令将 会 拉取并处理每个新事件，尽管最终会回滚更改。

错误处理

每个选定的跟随者都应有自己的同步机会。因此，该命令将逐个跟随者捕获一些异常，并继续处理剩余的跟随者。

捕获的基本Django异常包括 EmptyResultSet、FieldDoesNotExist、FieldError、MultipleObjectsReturned 和 ObjectDoesNotExist。来自 eventsourcing 库的基本异常 EventSourcingError 也按每个跟随者捕获。

配置

此命令需要访问一个 eventsourcing.system.Runner 实例以查询和对其跟随者采取行动。运行者的系统还定义了领导者和跟随者之间的管道。

默认行为是在不进行额外配置的情况下，检查所有已安装的Django应用，并查找 eventsourcing.system.Runner 的实例。属性名称不重要，只要它是公开的（即不以下划线开头）即可。

# djangoproject/apps/my_es_app/apps.py
import eventsourcing.system
from django.apps import AppConfig


class MyEventSourcedAppConfig(AppConfig):
   name = "my_event_sourced_app"
   runner: eventsourcing.system.Runner

   def ready(self) -> None:
       self.runner = eventsourcing.system.SingleThreadedRunner(
           eventsourcing.system.System(...)
       )

这通常足够了，除非您 i) 在一个或多个应用中定义了多个运行者，或 ii) 不在Django应用中持有运行者。在这种情况下，您应该在以下两种方式之一中配置Django设置 EVENTSOURCING_RUNNER：

1. 将 EVENTSOURCING_RUNNER 设置为应用名称的属性。此属性必须是 eventsourcing.system.Runner 实例。

   # djangoproject/settings.py
   ...
   EVENTSOURCING_RUNNER = "my_event_sourced_app.runner"
   

2. 将 EVENTSOURCING_RUNNER 设置为完全限定的函数名称。此函数将不带参数调用，并应返回一个 eventsourcing.system.Runner 实例。

   # djangoproject/settings.py
   ...
   EVENTSOURCING_RUNNER = "djangoproject.runner_utils.get_runner"
   

   # djangoproject/runner_utils.py
   import eventsourcing.system
   
   
   def get_runner() -> eventsourcing.system.Runner:
      return ...
   

eventsourcing 库中提供的所有运行者类都是兼容的。