Event-Sourced System

After you have defined your domain model, applications and projections, it is time to tie everything together. For that purpose, you use EventSourcedSystem class (referred to as the system), which orchestrates communication between applications and projections and executes commands and queries.

Firstly, the system is a boundary between your domain logic and the outside world. The outside world never calls use cases in your applications directly or retrieve data from projections; instead, you prepare commands and queries, and execute them using the system’s execute_command and execute_query methods. Such an approach allows you to hide the complexity of your business domain from the outside world behind the simple interface and gives you more flexibility to adapt your system for future changes.

Second, the system is responsible for the initialization and orchestration of applications and projections. You initialize them inside the system, and route commands and queries to the responsible handlers.

Defining the system

To create a system use EventSourcedSystem class:

from bestagon.core.system import EventSourcedSystem

class BondsSystem(EventSourcedSystem):
    def __init__(event_store: EventStore, checkpoint_store: CheckpointStore, neo_driver: AsyncDriver):
        super().__init__(event_store=event_store, checkpoint_store=checkpoint_store)

        self.repository = EventSourcedRepository(event_store=event_store)
        self.application = BondsApplication(repository=self.repository, checkpoint_store=checkpoint_store)
        self.bonds_projection = BondsProjection(checkpoint_store=checkpoint_store, driver=neo_driver)

The system takes several parameters as input:

  • event_store - it should be an instance of EventStore class to store events, generated by your model.

  • checkpoint_store - an instance of CheckpointStore class to store information about the last processed event for applications and projections.

  • Additional parameters, necessary for your system. In this case, it is an instance of AsyncDriver to access the neo4j database.

Next we create an instance of EventSourcedRepository class. It will be used by applications to save and retrieve aggregates. There is no use for multiple repositories, as in the traditional approach where you create a separate repository for each aggregate, event-sourcing allows us to greatly simplify this pattern and use a single repository for all aggregates.

After the repository, you define your applications. Here we have only one application - BondsApplication, which takes the repository and checkpoint store as parameters. Similarly, you add projections. In our example, there is only one projection - BondsProjection which takes the checkpoint store and neo4j driver as input parameters.

System initialization

It is not enough to define your applications and projections inside the system. To make it work, you need to initialize them and create subscriptions in the event store. For that purpose, you need to reimplement the initialize method and add your applications and projections to the system. To add applications, use the add_application method, to add projections, use the ‘add_projection’ method. Under the hood framework will make the necessary initializations and create subscriptions to the event store.

async def initialize(self) -> None:
    await super().initialize()
    await self.add_application(self.bonds_application)
    await self.add_projection(self.bonds_projection)

Routing commands and queries

The last thing to do is to register command and query handlers, so the system would know how exactly to execute commands and queries. For that purpose, you use an instance of Mapper class and use its register_command_handler and register_query_handler methods. One place to do that is inside register_command_handlers and register_query_handlers`of the `EventSourcedSystem class:

from bestagon.core.mapper import mapper

# Registering commands and quries inside the system
def register_command_handlers(self) -> None:
    mapper.register_command_handler(IssueBond, self.bonds_application.issue)
    mapper.register_commant_handler(UpdateTermToMaturity, self.bonds_aplication.update_term_to_maturity)
    mapper.register_command_handler(SetMatured, self.bonds_applicaiton.set_matured)

def register_query_handlers(self) -> None:
    mapperregister_query_handler(GetBondsQuery, self.bonds_projection.get_bonds)

After that, your system is ready to be used. In the next tutorial you will learn how to serve your system across the network using FastAPI.