Projection ========== In this tutorial you will learn how to create read models using projections. When you work with event-sourcing you store each event that has happened in your domain model in the event store and your event store is nothing more than a long chain of events. This approach is very powerful, because it gives you information about what has happened, when and what happened before, but it is very difficult to query such data structure to get necessary information for the decision-making process. As humans, we need all kinds of information to look at, analyze and make a decision. It can be a table, graph, chart and many other representations of data and to solve this problem we use read models. The projection is an implementation of a read model. It has no event store and never executes business logic. Its purpose is to react to events generated by a domain model and populate a read model, stored in a separate database. This database can be anything that suits your needs - PostgreSQL, Neo4J, Redis, MongoDB, in memory stores or your custom solutions. Depending on your requirements, you can create any number of projections you need, and each projection can be easily rebuilt or disposed of if you do not need it anymore, without any fear of losing critical information because this information is already stored in the event store as a set of immutable events. This gives you huge advantages: - Your system is no longer dependent on a specific technology, you can easily change databases as you need. For example, you can easily migrate from PostgreSQL to Neo4j without the need to rebuild all the system. - Projections are disposable, and you can easily build them, so you can quickly adapt to the new requirements for read models. Now let's take a look how to create and use projections. Creating abstract interface --------------------------- To create your own projection use `Projection` class. .. code-block:: python from bestagon.core.event_processor import Projection Before we move forward, it is good practice to define an abstract interface of your projection. It will make your life much easier if, at one point of time, you decide to switch the database behind the projection. In previous tutorials we have built an application for bonds and here we will build a projection that will give us an opportunity to query for bonds according to some criteria. .. code-block:: python from bestagon.core.event_processor import Projection class BondsProjection(Projection): # Add / Set methods are used to populate the read model, they return nothing @abstractmethod async def add_bond(self, isin: str, issue_date: str, maturity_date: str, coupon: float) -> None: raise NotImplementedError @abstractmethod async def set_term_to_maturity(self, isin: str, ttm: int) -> None: raise NotImplementedError @abstractmethod async def set_matured(self, isin: str, matured: bool) -> None: raise NotImplementedError # Query handler - you get information using this method @abstractmethod async def get_bonds(self, query: GetBondsQuery) -> List[dict]: raise NotImplementedError Query handlers -------------- This abstract interface will be used as a template for any future projection that stores information about bonds, and we can split these methods by two categories. The first are methods that receive input parameters to populate the read model. They usually start with `add` or `set` prefixes and return nothing. The second category is so-called query handlers. These methods take a Query as an input parameter and return requested data. The query is a subclass of `Query` class - it is a frozen dataclass with a set of predefined filters. In our example, we have one query - `GetBondsQuery` and it has only one attribute that allows us to filter bonds by whether they have matured or not. .. code-block:: python @dataclass(frozen=True) class GetBondsQuery(Query): matured: bool | None # Additional filters according to your requirements One thing to mention here is that query handler **must take a query** as an input parameter even if you do not need any filters. In such case, you create an empty query without filters and pass it as a parameter. Initialization -------------- Now it is time to write a projection. In this example we will use Neo4J as a database to store our read model: .. code-block:: python class Neo4JBondsProjection(BondsProjection): def __init__(checkpoint_store: CheckpointStore, driver: AsyncDriver): super().__init__(checkpoint_store) self.driver = driver self.database_name = 'bonds-projection' The projection takes several parameters as input. The first one is a checkpoint store instance. This parameter is mandatory, because, similar to `Application`, `Projection` needs to track which events have already been processed. All other parameters are related to the database you use to store your read model. In this example, it is a Neo4J driver that gives us acces to the database. In constructor, you should always call the constructor of the superclass, and then you define your projection-specific properties. When creating your projection, it is often required to configure the database you use. For example, create a separate database and indexes for the read model. To do that, you can use the `initialize` method, which will be called by the system at the startup stage: .. code-block:: python async def initialize(self) -> None: cypher = ''' CREATE DATABASE $database_name IF NOT EXISTS ''' async with self.driver.session() as sess: await sess.run(cypher, database_name=self.database_name) Also, similar to `Application` class you need to implement `get_name` method: .. code-block:: python def get_name(self) -> str: return 'bonds_projection' Populating projection --------------------- Now we are ready to start populating the projection. The process is very similar to what you saw in the application tutorial and consists of three steps: - Implement add/set methods we defined on abstract interface. - Provide even routing to map events to corresponding event handlers. - Implement event handlers to react to events. We start with implementation of add/set methods. In these methods, you receive input parameters and use them to populate the read model. Here is an example: .. code-block:: python async def add_bond(self, isin: str, issue_date: str, maturity_date: str, coupon: float) -> None: cypher = ''' MERGE (b:Bond {isin: $isin}) ON CREATE SET b += $props ''' props = { 'issue_date': date.fromisoformat(issue_date), 'maturity_date': date.fromisoformat(maturity_date), 'coupon': coupon } async with self.driver.session(default_access_mode=neo4j.WRITE_ACCESS, database=self.database_name) as sess: await sess.run(cypher, isin=isin, props=props) async def set_term_to_maturity(self, isin: str, ttm: int) -> None: cypher = ''' MATCH (b:Bond {isin: $isin}) SET b.ttm = $ttm ''' async with self.driver.session(default_access_mode=neo4j.WRITE_ACCESS, database=self.database_name) as sess: await sess.run(cypher, isin=isin, ttm=ttm) async def set_matured(self, isin: str, matured: bool) -> None: cypher = ''' MATCH (b:Bond {isin: $isin}) SET b.matured = $matured ''' async with self.driver.session(default_access_mode=neo4j.WRITE_ACCESS, database=self.database_name) as sess: await sess.run(cypher, isin=isin, matured=matured) These methods are used to populate the projection. Now we need to trigger them automatically as a reaction to specific events and the first thing we need to do is to reimplement the `get_event_routing` method to map events we want to react to, to the corresponding event handlers: .. code-block:: python def get_event_routing(self) -> dict: routing = { Bond.Issued: self._when_bond_issued, Bond.TermToMaturityChanged: self._when_ttm_changed, Bond.Matured: self._when_matured } return routing async def _when_bond_issued(self, event: Bond.Issued) -> None: await self.add_bond(isin=event.isin, issue_date=event.issue_date, maturity_date=event.maturity_date, coupon=event.coupon) async def _when_ttm_changed(self, event: Bond.TermToMaturityhanged) -> None: await self.set_term_to_maturity(isin=event.isin, ttm=event.new_ttm) -> None: async def _when_matured(self, event: Bond.Matured) -> None: await self.set_matured(isin=event.isin, matured=True) Here we also implement event handlers. The one and only responsibility of an event handler is to serve as a glue between an event and the method that populates the read model. The event handler takes an event as an input parameter and calls the necessary method that updates the read model. It should never contain any business logic and should not modify data in the database directly. Querying projection ------------------- Now we need to implement a query handler. This method is much more complex, because it contains logic to build a query, execute it and return the data. In this example we prepare a Cypher query with the necessary filters and retrieve data from the database as a list of dictionaries. This data can be easily passed through the network as a JSON file and converted to a table or any other visual representation: .. code-block:: python async def get_bonds(self, query: GetBondsQuery) -> List[dict]: filters = [ 'b.matured = $matured' if query.matured is not None else '', ] filters = [f for f in filters if f] if filters: filters = ' AND '.join(filters) filter_string = f'WHERE {filters}' else: filter_string = '' cypher = f''' MATCH (b:Bond) {filter_string} WITH {{ isin: b.isin, issue_date: toString(b.issue_date), maturity_date: toString(b.maturity_date), coupon: b.coupon, ttm: b.ttm, matured: coalesce(b.matured, false) }} AS datum RETURN datum ''' async with self.driver.session(default_access_mode=neo4j.READ_ACCESS, database=self.database_name) as sess: result = await sess.run(cypher, matured=query.matured, bond_type=query.bond_type) data = await result.value() return data Dropping projection ------------------- Projections are disposable and to be able to comply with this requirement you need to reimplement an abstract `drop` method. Here you need to implement a logic to remove all the data from the database regarding the projection and should not bother about resetting checkpoints, because it is the responsibility of `EventSourcedSystem` class. In this example we implement a simple logic - we replace the existing database with the new one, removing all the data in it: .. code-block:: python async def drop(self) -> None: cypher = 'CREAE OR REPLACE DATBASE $database_name' async with self.driver.session() as sess: await sess.run(cypher, database_name=self.database) After that, your projection is complete. Remember - you can define any number of projections you want using any storage that is available for you. Projectors can be easily dropped and rebuilt again without the fear of losing important data, because they are built from the source of truth - events. In the next tutorial you will learn how to tie everything together and create an even-sourced system.