Configuring Event Storage with PostgreSQL

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

In the last lesson, we built a BankingApplication using the core abstractions of the eventsourcing library: aggregates, events, and application services. Our application correctly enforces business logic and records state changes as a sequence of events. However, it has a critical limitation: it uses the library's default in-memory persistence. This means all data is lost the moment the application stops.

Today's lesson directly addresses this by tackling the learning outcome:

Configure the framework to use a separate persistence service (e.g., PostgreSQL) for event storage.

We will take the exact BankingApplication we built and, with only configuration changes, make it durable by persisting its events to a PostgreSQL database. This will powerfully illustrate the separation of concerns between domain logic and infrastructure that is a key benefit of this architectural style.

1. Setting Up a PostgreSQL Service

Before our Python application can talk to a database, we need a database server running. We will use PostgreSQL, a robust and popular open-source relational database.

For local development, the most common and reproducible way to run services like PostgreSQL is with Docker. If you don't have it installed, you can find instructions at the official Docker website.

Here is a docker-compose.yml file that defines a PostgreSQL service. Create a file with this name and content in your project directory.

# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: postgres:15
    container_name: es-postgres
    environment:
      POSTGRES_DB: eventsourcing
      POSTGRES_USER: eventsourcing
      POSTGRES_PASSWORD: eventsourcing
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

This configuration starts a PostgreSQL 15 container, creates a database named eventsourcing with a corresponding user and password, and maps the container's port 5432 to your local machine's port 5432.

To start the service, run the following command in your terminal from the same directory as the file:
docker-compose up -d

The -d flag runs the container in detached mode. You can verify it's running with docker ps.

If you prefer a manual installation or want to understand the basics of interacting with PostgreSQL from Python, the following video provides a good overview. However, for this lesson, all you need is the Docker container running.

2. The eventsourcing Library's Persistence Abstractions

The eventsourcing library is designed to be persistence-agnostic. It achieves this through an adapter-like pattern centered on an object called a Recorder. A recorder's job is to translate the library's internal representation of events into the specific commands required by a database system (e.g., SQL INSERT statements for PostgreSQL).

The library provides different recorders for different needs and databases.

3. Configuration-Driven Persistence

While you can manually construct the persistence objects as shown in the documentation, the framework offers a much more flexible and conventional approach: configuration via environment variables.

The library's Application class (which our BankingApplication inherits from) uses an InfrastructureFactory behind the scenes. When an application instance is created, this factory reads environment variables to determine which persistence backend to use and how to connect to it.

This is extremely powerful. It means we can switch our application from in-memory to a production-grade PostgreSQL database without changing a single line of our domain or application code.

4. Practical Application: Making the BankingApplication Durable

Now, let's put this all together. We will use the BankingApplication from the previous lesson and run it with the environment configured for PostgreSQL.

First, here is the complete code for our application, including the Account aggregate and BankingApplication service. Save this as banking.py.

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

# 1. Aggregate
class InsufficientFundsError(Exception):
    pass

class Account(Aggregate):
    """A bank 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):
        """Class method to create a new account."""
        return cls(owner=owner, initial_balance=initial_balance)

    @event('FundsDeposited')
    def deposit(self, amount: Decimal):
        if self.is_closed:
            raise ValueError("Account is closed.")
        if amount <= 0:
            raise ValueError("Deposit amount must be positive.")
        self.balance += amount

    @event('FundsWithdrawn')
    def withdraw(self, amount: Decimal):
        if self.is_closed:
            raise ValueError("Account is closed.")
        if amount <= 0:
            raise ValueError("Withdrawal amount must be positive.")
        if self.balance < amount:
            raise InsufficientFundsError("Insufficient funds.")
        self.balance -= amount

    @event('AccountClosed')
    def close(self):
        if self.is_closed:
            raise ValueError("Account is already closed.")
        if self.balance != Decimal('0'):
            raise ValueError("Cannot close account with a non-zero balance.")
        self.is_closed = True

# 2. Application
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)
        
    def close_account(self, account_id: uuid.UUID):
        account: Account = self.repository.get(account_id)
        account.close()
        self.save(account)

    def get_balance(self, account_id: uuid.UUID) -> Decimal:
        account: Account = self.repository.get(account_id)
        return account.balance

# 3. Runner script
if __name__ == '__main__':
    # Set environment variables for PostgreSQL persistence
    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' # Create tables if they don't exist

    app = BankingApplication()

    # Create an account and get its ID
    account_id = app.open_account(owner="Alice", initial_balance=Decimal('100.00'))
    print(f"Opened account for Alice with ID: {account_id}")

    # Perform some transactions
    app.deposit_funds(account_id, Decimal('50.00'))
    app.withdraw_funds(account_id, Decimal('25.00'))

    # Check the balance
    balance = app.get_balance(account_id)
    print(f"Current balance for Alice's account: {balance}")

    print("\nApplication finished. State is persisted in PostgreSQL.")
    print("Run verify_persistence.py to see the state loaded from the DB.")

Next, create a second file, verify_persistence.py, to prove that the state was saved.

# verify_persistence.py
import os
import uuid
from decimal import Decimal
from banking import BankingApplication

if __name__ == '__main__':
    # Configure the environment to use the same PostgreSQL database
    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'
    # We can set CREATE_TABLE to 'n' here, as we expect the table to exist
    os.environ['CREATE_TABLE'] = 'n' 

    app = BankingApplication()

    # Prompt user for the account ID from the previous run
    account_id_str = input("Enter the account ID for Alice: ")
    try:
        account_id = uuid.UUID(account_id_str)
    except ValueError:
        print("Invalid UUID format.")
        exit()

    # Retrieve the balance. This will replay events from the database.
    try:
        balance = app.get_balance(account_id)
        print(f"\nSuccessfully retrieved account {account_id}.")
        print(f"Current balance from database: {balance}")
        assert balance == Decimal('125.00')
        print("Balance is correct. Persistence is working!")
    except Exception as e:
        print(f"Could not retrieve account. Error: {e}")

Exercise:

1. Make sure your PostgreSQL container is running (docker-compose up -d).
2. Run the first script: python banking.py. Note the account ID that is printed.
3. Run the second script: python verify_persistence.py.
4. When prompted, paste the account ID from the first run.

You should see the script successfully retrieve the correct balance of 125.00, confirming that the account's state was loaded by replaying the events stored in PostgreSQL.

To see the raw data, you can connect to the database. One way is to use psql inside the Docker container:

# Get a shell inside the container
docker exec -it es-postgres bash

# Connect to the database with psql
psql -U eventsourcing -d eventsourcing

# Once inside psql, run this query to see the events table:
SELECT * FROM stored_events;

You will see rows corresponding to the Opened, FundsDeposited, and FundsWithdrawn events, with the event data stored as a JSON blob.

Conclusion

In this lesson, you successfully configured the eventsourcing framework to use an external, durable persistence service. The most important takeaway is how this was achieved purely through configuration, demonstrating a clean separation between your application's core logic and its infrastructure dependencies.

Key Takeaways:

• The eventsourcing library uses Recorders as adapters to communicate with different database backends.
• The InfrastructureFactory reads environment variables to automatically construct and configure the necessary persistence components.
• By setting PERSISTENCE_MODULE to eventsourcing.postgres and providing connection details, you can switch from in-memory storage to PostgreSQL without any code changes in your aggregates or application services.
• The CREATE_TABLE='y' setting is a convenient feature for development, as it instructs the framework to create the necessary database tables if they don't already exist.

Preview of the Next Lesson:

Now that we have a durable, append-only log of events in our database, we can leverage it for more than just reconstructing aggregates. The write-side is optimized for consistency, but the read-side often has different requirements (e.g., complex queries, joins, high performance). In the next lesson, "Implement projections and read models that are driven by the framework's notification system," we will explore how to create optimized read models by consuming the event stream, completing the Command Query Responsibility Segregation (CQRS) pattern.