Application

In this tutorial we will take a look at event-sourced application in more detail. When you have your domain model implemented with event-sourced aggregates, you need to somehow trigger the business logic inside them and interact with other aggregates. Despite the fact that you can use aggregates directly, it is not advised. Usually, you have a set of use cases in your business which start a process that involves one or multiple aggregates and implementation of these use cases is where an application steps in. For that purpose, you use Application class.

from bestagon.core.event_processor import Applicaiton

An application serves multiple purposes:

It has its own domain model.
It includes use cases to work with domain model.
It contains a persistence mechanism, able to store and retrieve event-sourced aggregates.
It contains a mechanism to react to events that have been produced by your domain model.

You can think of an application like a microservice inside a microservice - it is a boundary with strict borders where each application has its own domain model, knows nothing about other applications, but can react to events generated by other domain models.

Application name

Before we dive into implementation of use cases and reaction to events, we should address some technical details regarding the Application class. The first one is get_name method. Similar to Aggregate which should have a unique ID, your applications should also have a unique identifier - its name. It is used internally by the system and should not be changed after you define it, because it can corrupt your system.

class BondsApplication(Application):

    def get_name(self) -> str:
        return 'bonds_application'  # Unique across your system

Event-sourced repository

These are technical details for those who want to get a better understanding of how the framework works internally. In most cases, you do not need to reimplement the constructor of the application, and you can concentrate on writing business logic.

When you initialize your application, it takes two arguments as input - repository and checkpoint_store`:

class BondsApplication(Application):

    def __init__(self, repository: EventSourcedRepository, checkpoint_store: CheckpointStore):
        # Initialization logic is here

Repository is a subclass of EventSourcedRepository class. This component plays a crucial part in the application and implements a persistence mechanism inside your application that allows you to save and retrieve aggregates. It hides the event store and persistence mechanism behind the simple interface:

get_by_id method allows you to retreive stored aggregate from event store.
save method saves the aggregate in event store.
contains method allows you to check whether aggregate exists in repository.

IMPORTANT - all these methods are asynchronous and should be used with async / await syntax.

You can access repository inside your application at any time using repository property.

Checkpoint store

Another input parameter is a checkpoint_store. Each application contains a subscription to the event store, reads events from it one by one and reacts in a specific way. Such an approach allows you to automatically react to events and automate the execution of use cases, but it requires a mechanism to save information which events have already been processed.

For that purpose, a checkpoint store has been designed. It is a subclass of the abstract CheckpointStore class with a simple interface. It simply stores the position of the last processed event in the event store, and if the system is restarted or crashed, it will be able to continue its work after the last processed event.

WARNING - checkpoint store uses the application name as a key to the checkpoint and if you change the application name after you have already started to process events it will break your application because it will not be able to find the checkpoint for that name, will create a new one and start to process events that have already been processed.

Defining use cases

Now it is time to write use cases. Similar to Aggregate there are two types of methods - command handlers and event handlers.

Command handlers are your actual use cases - public methods with names that should reflect your business domain, and they receive one and only one parameter as input - a command. Inside a command handler, you create new aggregates, retrieve already existing from the repository, execute business logic inside aggregates and orchestrate communication between multiple aggregates. Command handlers return nothing, their responsibility is to execute business logic and save changes in the aggregates in the repository. Here is a simple example - for our application to work we need to be able to issue new bonds:

class BondsApplication(Application):

    # Command handler name reflects you business domain and it receives a Command as input parameter
    async def issue_bond(self, command: IssueBond) -> None:
        # Here you can add a validation logic for your command values
        # If the data inside the command is corrupted, then use case should fail with exception

        # Before creating new aggregate we check for duplicates and return if this bond have already been issued
        aggregate_id = Bond.create_id(isin=command.isin)
        if await self.repository.contains(aggregate_type=Bond.get_aggregate_type(), aggregate_id=aggregate_id):
            logger.info(f'Duplicate bond {command.isin}')
            return

        # Create new aggregate and save it in repository
        bond = Bond.issue(command)
        await self.repository.save(bond)

NOTE - command handlers are asynchronous functions.

IMPORTANT - when you write your use cases, it is highly advisable to stick to the rule of thumb - modify only one aggregate at a time. Situations when you need to modify multiple aggregates at once show you weak points in your design, and it is a signal that you need to return to the design of your domain model.

In addition to command handlers, it is good practice to add getters to your application to retrieve aggregates from repository, this will allow you to concentrate on your business logic instead of the technical details of a framework.

async def get_bond(self, isin: str) -> Bond:
    aggregate_id = Bond.create_id(isin)
    bond = await self.repository.get_by_id(aggregate_type=Bond.get_aggregate_type(), aggregate_id=aggregate_id)
    return bond

Let’s add another couple of use cases to the application. We need to be able to update term to maturity of the bond and set the bond as matured:

async def set_matured(self, command: SetMatured) -> None:
    bond = await self.get_bond(isin=command.isin)
    if bond.ready_to_mature():
        bond.set_matured()
        await self.repository.save(bond)

async def update_term_to_maturity(self, command: UpodateTermToMaturity) -> None:
    # First get bond from repository
    bond = await self.get_bond(isin=command.isin)

    # Execute business logic in your aggregate
    bond.update_term_to_maturity()

    # Save changes in repository
    await self.repository.save(bond)

Reaction to events

Bestagon provides a mechanism of reaction to the events generated by applications. It allows you to trigger use cases automatically as a reaction to a specific event. To benefit from this mechanism, you need three things: command handler, event handler and event routing.

We have already defined several command handlers for our application. Now let’s take a look at event routing. An event routing is a simple dictionary where the keys are event types you want to react to and values are corresponding event handlers. To provide event routing for your application, you need to reimplement get_event_routing abstract method:

def get_event_routing() -> dict:
    routing = {
        Bond.Issued: self._when_bond_issued,
        Bond.TermToMaturityChanged: self._when_term_to_maturity_changed,
    }
    return routing

Here we provide our application with reactions to the events - what happens when the bond is issued and its term to maturity changed.

IMPORTANT - here we react to local events that have been generated by domain model in the application, but you can also react to events from other applications simply by adding them to event routing.

The private methods in the event routing are event handlers. They are asynchronous methods that receive an event as an input parameter and trigger a specific use case. You do not implement business logic inside an event handler. Its main goal is to take an event and trigger a corresponding use case, so it is good practice to make them as thin as possible:

async def _when_bond_issued(self, event: Bond.Issued) -> None:
    command = UpdateTermToMaturity(isin=event.isin)
    await self.update_term_to_maturity(command)

async def _when_term_to_maturity_changed(self, event: Bond.TermToMaturityChanged) -> None:
    command = SetMatured(isin=event.isin)
    await self.set_matured(command)

Now you know how to write use cases and react to events using Application class. In the next tutorial you will learn how Bestagon implements CQRS pattern to create read models with Projection class.