Building Projections and Read Models with Notifications

Hello! Welcome back to our course on distributed systems architecture.

In our last lesson, you successfully configured our BankingApplication to persist its event stream to a PostgreSQL database. This gave our write-model durability. However, the event stream, while perfect for ensuring consistency and providing an audit log, is not optimized for complex queries. Retrieving an aggregate by its ID is efficient, but answering questions like "What are all the accounts owned by Alice?" or "Show me all accounts with a balance over £10,000" would require replaying every event for every account, which is highly impractical.

This lesson addresses that exact problem by introducing the read-model side of the CQRS pattern. We will tackle the following learning outcome:

Implement projections and read models that are driven by the framework's notification system.

We will build a separate, query-optimized "read model" that provides a simple list of all bank accounts. This read model will be populated by a "projection" process that listens to the stream of events from our BankingApplication and updates its own state accordingly. This demonstrates the full power of separating write and read responsibilities in an event-sourced system.

1. Conceptual Foundation: Projections and Read Models

Before diving into the code, it's crucial to understand the concepts of Command Query Responsibility Segregation (CQRS), projections, and read models. In an event-sourced system, the event store is the write model—the single source of truth. A read model (or "materialized view") is a denormalized representation of data tailored for a specific query use case. A projection is the process that creates and updates a read model by consuming events from the write model's event stream.

One event can feed multiple projections, each creating a different read model for a different purpose (e.g., one for a customer-facing list, another for internal risk analysis).

Given your background, you can think of a projection as being analogous to a materialized view in a relational database. A materialized view pre-computes and stores the result of a query. Similarly, a projection pre-computes a queryable state from a stream of events.

2. The eventsourcing Library's Projection Framework

The eventsourcing library provides a set of classes to formalize the implementation of projections and read models.

To summarize the key components we will build:

1. Read Model (Materialized View): A class that inherits from PostgresTrackingRecorder. It will manage its own database table(s) to store the queryable state and will use the tracking mechanism from the parent class to ensure events are processed exactly once.
2. Projection: A class that inherits from Projection. It will contain the business logic, implemented as event handler methods, that decides how to update the read model in response to specific events.
3. Runner Script: An executable script that configures and starts a ProjectionRunner, linking our BankingApplication (the event source) with our new projection and read model.

3. Practical Implementation: An "All Accounts" Read Model

Let's build a read model that maintains a simple table of all accounts, showing their ID, owner, and current balance. This will allow us to query for a list of all accounts, a feature not easily supported by the write model.

Step 1: Define the Read Model

First, we define the class that will manage our read model's data in PostgreSQL. This class needs to handle creating the table, updating rows, and providing a query method.

Create a new file, projections.py, and add the following code.

# projections.py
import uuid
from abc import ABC, abstractmethod
from decimal import Decimal
from typing import List, Dict, Any

from eventsourcing.dispatch import singledispatchmethod
from eventsourcing.persistence import Tracking
from eventsourcing.postgres import PostgresTrackingRecorder
from eventsourcing.projection import Projection
from psycopg2.extras import RealDictCursor

# Import events from our banking application
# (We will create banking_app.py in the next step)
from banking_app import Account, InsufficientFundsError

# Define an interface for our read model for type hinting and clarity
class IAccountsView(ABC):
    @abstractmethod
    def get_accounts(self) -> List[Dict[str, Any]]:
        pass

    @abstractmethod
    def upsert_account(self, account_id: uuid.UUID, owner: str, balance: Decimal, tracking: Tracking):
        pass

# Implement the read model using PostgreSQL
class AccountsView(PostgresTrackingRecorder, IAccountsView):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.accounts_table = self.tracking_table_name.replace("_tracking", "_accounts")

        # SQL to create the accounts table
        self.sql_create_statements.append(
            f"""
            CREATE TABLE IF NOT EXISTS {self.accounts_table} (
                account_id UUID PRIMARY KEY,
                owner VARCHAR NOT NULL,
                balance DECIMAL NOT NULL
            )
            """
        )

    def get_accounts(self) -> List[Dict[str, Any]]:
        with self.datastore.transaction(commit=False) as curs:
            curs.execute(f"SELECT account_id, owner, balance FROM {self.accounts_table}")
            # RealDictCursor returns rows as dictionaries
            return curs.fetchall()

    def upsert_account(self, account_id: uuid.UUID, owner: str, balance: Decimal, tracking: Tracking):
        # The 'with self.datastore.transaction()' block ensures that updating the read model
        # and updating the tracking record happen atomically. If any part fails,
        # the whole transaction is rolled back.
        with self.datastore.transaction(commit=True) as curs:
            # First, record that we are processing this event.
            # This prevents double-processing if the service restarts.
            self._insert_tracking(curs, tracking)

            # Then, update the read model table
            sql = f"""
                INSERT INTO {self.accounts_table} (account_id, owner, balance)
                VALUES (%(account_id)s, %(owner)s, %(balance)s)
                ON CONFLICT (account_id) DO UPDATE SET
                    owner = EXCLUDED.owner,
                    balance = EXCLUDED.balance
            """
            params = {
                "account_id": account_id,
                "owner": owner,
                "balance": balance,
            }
            curs.execute(sql, params)

This AccountsView class does two critical things:

1. It defines and creates a new _accounts table to store the denormalized view.
2. Its upsert_account method wraps the database update and the call to _insert_tracking in a single atomic transaction. This is essential for reliability.

Step 2: Define the Projection Logic

Next, we define the Projection subclass that translates events into calls on our AccountsView.

Add this code to the bottom of projections.py.

# projections.py (continued)

class AccountsProjection(Projection[AccountsView]):
    # The name is used to prefix environment variables for configuration
    name = "accounts"
    
    # We are interested in all events related to the Account aggregate
    topics = (
        "banking_app.Account",
    )

    @singledispatchmethod
    def process_event(self, domain_event: object, tracking: Tracking) -> None:
        """Default event handler, ignores events we don't know about."""
        self.view.insert_tracking(tracking)

    @process_event.register
    def _(self, event: Account.Opened, tracking: Tracking) -> None:
        # When an account is opened, create a new record in the view.
        self.view.upsert_account(
            account_id=event.originator_id,
            owner=event.owner,
            balance=event.initial_balance,
            tracking=tracking,
        )

    @process_event.register
    def _(self, event: Account.FundsDeposited, tracking: Tracking) -> None:
        # When funds are deposited, update the balance.
        self.view.upsert_account(
            account_id=event.originator_id,
            owner=event.owner, # Note: we need to pass all fields for the upsert
            balance=event.balance,
            tracking=tracking,
        )

    @process_event.register
    def _(self, event: Account.FundsWithdrawn, tracking: Tracking) -> None:
        # When funds are withdrawn, update the balance.
        self.view.upsert_account(
            account_id=event.originator_id,
            owner=event.owner,
            balance=event.balance,
            tracking=tracking,
        )
    
    # We don't need to handle AccountClosed explicitly for this view,
    # but in a real system you might want to remove the account from the list
    # or mark it as closed.

Note the use of @singledispatchmethod. This is a clean way to route events to different handler methods based on their type. Each handler extracts the relevant data from the event and calls the upsert_account method on the view.

4. Running and Querying the System

Now we need three scripts to tie everything together:

1. banking_app.py: The write-model code from the previous lesson.
2. run_projection.py: A script to start and run our projection process.
3. client.py: A script to interact with the system—both writing commands and reading from our new view.

1. The Write Model (banking_app.py)

Save the application code from our previous lesson into a file named banking_app.py. For clarity, I've reproduced it here.

# banking_app.py
import uuid
from decimal import Decimal
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate, event

class InsufficientFundsError(Exception):
    pass

class Account(Aggregate):
    def __init__(self, owner: str, initial_balance: Decimal = Decimal('0')):
        self.owner = owner
        self.balance = initial_balance
        self.is_closed = False

    @event('Opened')
    @classmethod
    def open(cls, owner: str, initial_balance: Decimal):
        return cls(owner=owner, initial_balance=initial_balance)

    @event('FundsDeposited')
    def deposit(self, amount: Decimal):
        self.balance += amount

    @event('FundsWithdrawn')
    def withdraw(self, amount: Decimal):
        if self.balance < amount:
            raise InsufficientFundsError("Insufficient funds.")
        self.balance -= amount

    @event('AccountClosed')
    def close(self):
        self.is_closed = True

class BankingApplication(Application):
    def open_account(self, owner: str, initial_balance: Decimal) -> uuid.UUID:
        account = Account.open(owner=owner, initial_balance=initial_balance)
        self.save(account)
        return account.id

    def deposit_funds(self, account_id: uuid.UUID, amount: Decimal):
        account: Account = self.repository.get(account_id)
        account.deposit(amount)
        self.save(account)

    def withdraw_funds(self, account_id: uuid.UUID, amount: Decimal):
        account: Account = self.repository.get(account_id)
        account.withdraw(amount)
        self.save(account)

2. The Projection Runner (run_projection.py)

This script will run in its own terminal. It does nothing but listen for new events and update the read model.

# run_projection.py
import os
from eventsourcing.projection import ProjectionRunner
from banking_app import BankingApplication
from projections import AccountsProjection, AccountsView

# Set environment variables for both the application and the projection
# The 'APPLICATION_' prefix is used by the BankingApplication
# The 'ACCOUNTS_' prefix is used by our AccountsProjection (because its name is 'accounts')
os.environ['PERSISTENCE_MODULE'] = 'eventsourcing.postgres'
os.environ['POSTGRES_HOST'] = '127.0.0.1'
os.environ['POSTGRES_DBNAME'] = 'eventsourcing'
os.environ['POSTGRES_USER'] = 'eventsourcing'
os.environ['POSTGRES_PASSWORD'] = 'eventsourcing'
os.environ['CREATE_TABLE'] = 'y'

# The projection needs its own config, prefixed with its name
os.environ['ACCOUNTS_PERSISTENCE_MODULE'] = 'eventsourcing.postgres'
os.environ['ACCOUNTS_POSTGRES_HOST'] = '127.0.0.1'
os.environ['ACCOUNTS_POSTGRES_DBNAME'] = 'eventsourcing'
os.environ['ACCOUNTS_POSTGRES_USER'] = 'eventsourcing'
os.environ['ACCOUNTS_POSTGRES_PASSWORD'] = 'eventsourcing'
os.environ['ACCOUNTS_CREATE_TABLE'] = 'y'

if __name__ == '__main__':
    with ProjectionRunner(
        application_class=BankingApplication,
        projection_class=AccountsProjection,
        view_class=AccountsView,
    ) as runner:
        print("Projection runner started. Press Ctrl+C to exit.")
        runner.run_forever()

3. The Client (client.py)

This script allows us to create accounts (write side) and view the list of all accounts (read side).

# client.py
import os
from decimal import Decimal
from pprint import pprint

from eventsourcing.persistence import InfrastructureFactory

from banking_app import BankingApplication
from projections import AccountsView, IAccountsView

# Set environment variables
os.environ['PERSISTENCE_MODULE'] = 'eventsourcing.postgres'
os.environ['POSTGRES_HOST'] = '127.0.0.1'
os.environ['POSTGRES_DBNAME'] = 'eventsourcing'
os.environ['POSTGRES_USER'] = 'eventsourcing'
os.environ['POSTGRES_PASSWORD'] = 'eventsourcing'

os.environ['ACCOUNTS_PERSISTENCE_MODULE'] = 'eventsourcing.postgres'
os.environ['ACCOUNTS_POSTGRES_HOST'] = '127.0.0.1'
os.environ['ACCOUNTS_POSTGRES_DBNAME'] = 'eventsourcing'
os.environ['ACCOUNTS_POSTGRES_USER'] = 'eventsourcing'
os.environ['ACCOUNTS_POSTGRES_PASSWORD'] = 'eventsourcing'

if __name__ == '__main__':
    # Get the write model
    write_model = BankingApplication()

    # Get the read model
    # We use the InfrastructureFactory to construct the view directly
    read_model: IAccountsView = InfrastructureFactory.construct(
        {'PERSISTENCE_MODULE': os.getenv('ACCOUNTS_PERSISTENCE_MODULE')}
    ).tracking_recorder(AccountsView)

    while True:
        print("\nOptions:")
        print("1. Open new account")
        print("2. Show all accounts (from read model)")
        print("3. Exit")
        choice = input("Choose an option: ")

        if choice == '1':
            owner = input("Enter owner's name: ")
            balance = input("Enter initial balance: ")
            try:
                account_id = write_model.open_account(owner, Decimal(balance))
                print(f"Successfully opened account {account_id} for {owner}.")
                print("Read model will update shortly (eventual consistency).")
            except Exception as e:
                print(f"Error: {e}")
        
        elif choice == '2':
            print("\n--- All Accounts (Read Model) ---")
            accounts = read_model.get_accounts()
            if not accounts:
                print("No accounts found.")
            else:
                pprint(accounts)
            print("---------------------------------")

        elif choice == '3':
            break

---

Exercise

1. Make sure your PostgreSQL Docker container is running (docker-compose up -d).
2. Open a terminal and run the projection: python run_projection.py. Leave this running.
3. Open a second terminal and run the client: python client.py.
4. In the client terminal, choose option 2 to show all accounts. It should be empty.
5. Now, choose option 1 to open a few accounts (e.g., for 'Alice', 'Bob', 'Charlie').
6. After creating them, choose option 2 again. You should now see the accounts listed, demonstrating that the projection has processed the events and updated the read model.
7. You can also inspect the database directly to see the new _accounts table:

   docker exec -it es-postgres psql -U eventsourcing -d eventsourcing
   \dt # list tables
   SELECT * FROM accounts_projection_accounts; # table name is <projection_name>_projection_<view_table_name>
   

---

Conclusion

In this lesson, you have successfully implemented a core pattern in event-driven architecture. By creating a projection and a read model, you have decoupled the system's write and read concerns, enabling you to build query-optimized views of your data without compromising the integrity of your event-sourced write model.

Key Takeaways:

• Projections are processes that consume an event stream to build and maintain read models.
• Read models are denormalized, query-optimized data structures tailored to specific application needs.
• The eventsourcing library provides Projection, TrackingRecorder, and ProjectionRunner classes to structure this pattern.
• The ProjectionRunner uses the application's notification system (e.g., PostgreSQL's NOTIFY/LISTEN) to drive the projection in near real-time.
• Wrapping read model updates and tracking record inserts in a single database transaction is key to building a reliable, "exactly-once" processing system.

Preview of the Next Lesson:

Our system is now significantly more complex, with a write application and a separate read-side projection process. How can we be confident that all these moving parts work correctly together? In the next lesson, "Write unit and integration tests for an event-sourced application using the framework's testing utilities," we will explore strategies for testing aggregates, application services, and our newly created projections.