Application 1 - Bank accounts

This example demonstrates a straightforward event-sourced application.

Application

The BankAccounts application has command and query methods for interacting with the domain model. New accounts can be opened. Existing accounts can be closed. Deposits and withdraws can be made on open accounts. Transfers can be made between open accounts, if there are sufficient funds on the debited account. All actions are atomic, including transfers between accounts.

from __future__ import annotations

from decimal import Decimal
from uuid import UUID

from eventsourcing.application import AggregateNotFound, Application
from eventsourcing.examples.bankaccounts.domainmodel import BankAccount


class BankAccounts(Application):
    def open_account(self, full_name: str, email_address: str) -> UUID:
        account = BankAccount(
            full_name=full_name,
            email_address=email_address,
        )
        self.save(account)
        return account.id

    def get_account(self, account_id: UUID) -> BankAccount:
        try:
            return self.repository.get(account_id)
        except AggregateNotFound:
            raise AccountNotFoundError(account_id)

    def get_balance(self, account_id: UUID) -> Decimal:
        account = self.get_account(account_id)
        return account.balance

    def deposit_funds(self, credit_account_id: UUID, amount: Decimal) -> None:
        account = self.get_account(credit_account_id)
        account.credit(amount)
        self.save(account)

    def withdraw_funds(self, debit_account_id: UUID, amount: Decimal) -> None:
        account = self.get_account(debit_account_id)
        account.debit(amount)
        self.save(account)

    def transfer_funds(
        self,
        debit_account_id: UUID,
        credit_account_id: UUID,
        amount: Decimal,
    ) -> None:
        debit_account = self.get_account(debit_account_id)
        credit_account = self.get_account(credit_account_id)
        debit_account.debit(amount)
        credit_account.credit(amount)
        self.save(debit_account, credit_account)

    def set_overdraft_limit(self, account_id: UUID, overdraft_limit: Decimal) -> None:
        account = self.get_account(account_id)
        account.set_overdraft_limit(overdraft_limit)
        self.save(account)

    def get_overdraft_limit(self, account_id: UUID) -> Decimal:
        account = self.get_account(account_id)
        return account.overdraft_limit

    def close_account(self, account_id: UUID) -> None:
        account = self.get_account(account_id)
        account.close()
        self.save(account)


class AccountNotFoundError(Exception):
    pass

Domain model

The BankAccount aggregate class is defined using the declarative syntax. It has a balance and an overdraft limit. Accounts can be opened and closed. Accounts can be credited and debited, which affects the balance. Neither credits nor debits are allowed if the account has been closed. Debits are not allowed if the balance would go below the overdraft limit. The overdraft limit can be adjusted.

from __future__ import annotations

from decimal import Decimal

from eventsourcing.domain import Aggregate, event


class BankAccount(Aggregate):
    @event("Opened")
    def __init__(self, full_name: str, email_address: str):
        self.full_name = full_name
        self.email_address = email_address
        self.balance = Decimal("0.00")
        self.overdraft_limit = Decimal("0.00")
        self.is_closed = False

    @event("Credited")
    def credit(self, amount: Decimal) -> None:
        self.check_account_is_not_closed()
        self.balance += amount

    @event("Debited")
    def debit(self, amount: Decimal) -> None:
        self.check_account_is_not_closed()
        self.check_has_sufficient_funds(amount)
        self.balance -= amount

    @event("OverdraftLimitSet")
    def set_overdraft_limit(self, overdraft_limit: Decimal) -> None:
        assert overdraft_limit > Decimal("0.00")
        self.check_account_is_not_closed()
        self.overdraft_limit = overdraft_limit

    @event("Closed")
    def close(self) -> None:
        self.is_closed = True

    def check_account_is_not_closed(self) -> None:
        if self.is_closed:
            raise AccountClosedError({"account_id": self.id})

    def check_has_sufficient_funds(self, amount: Decimal) -> None:
        if self.balance - amount < -self.overdraft_limit:
            raise InsufficientFundsError({"account_id": self.id})


class TransactionError(Exception):
    pass


class AccountClosedError(TransactionError):
    pass


class InsufficientFundsError(TransactionError):
    pass

Test case

For the purpose of showing how the application object might be used, the test runs through a scenario that exercises all the methods of the application in one test method.

from __future__ import annotations

import unittest
from decimal import Decimal
from uuid import uuid4

from eventsourcing.examples.bankaccounts.application import (
    AccountNotFoundError,
    BankAccounts,
)
from eventsourcing.examples.bankaccounts.domainmodel import (
    AccountClosedError,
    InsufficientFundsError,
)


class TestBankAccounts(unittest.TestCase):
    def test(self) -> None:
        app = BankAccounts()

        # Check account not found error.
        with self.assertRaises(AccountNotFoundError):
            app.get_balance(uuid4())

        # Create account #1.
        account_id1 = app.open_account(
            full_name="Alice",
            email_address="alice@example.com",
        )

        # Check balance of account #1.
        self.assertEqual(app.get_balance(account_id1), Decimal("0.00"))

        # Deposit funds in account #1.
        app.deposit_funds(
            credit_account_id=account_id1,
            amount=Decimal("200.00"),
        )

        # Check balance of account #1.
        self.assertEqual(app.get_balance(account_id1), Decimal("200.00"))

        # Withdraw funds from account #1.
        app.withdraw_funds(
            debit_account_id=account_id1,
            amount=Decimal("50.00"),
        )

        # Check balance of account #1.
        self.assertEqual(app.get_balance(account_id1), Decimal("150.00"))

        # Fail to withdraw funds from account #1- insufficient funds.
        with self.assertRaises(InsufficientFundsError):
            app.withdraw_funds(
                debit_account_id=account_id1,
                amount=Decimal("151.00"),
            )

        # Check balance of account #1 - should be unchanged.
        self.assertEqual(app.get_balance(account_id1), Decimal("150.00"))

        # Create account #2.
        account_id2 = app.open_account(
            full_name="Bob",
            email_address="bob@example.com",
        )

        # Transfer funds from account #1 to account #2.
        app.transfer_funds(
            debit_account_id=account_id1,
            credit_account_id=account_id2,
            amount=Decimal("100.00"),
        )

        # Check balances.
        self.assertEqual(app.get_balance(account_id1), Decimal("50.00"))
        self.assertEqual(app.get_balance(account_id2), Decimal("100.00"))

        # Fail to transfer funds - insufficient funds.
        with self.assertRaises(InsufficientFundsError):
            app.transfer_funds(
                debit_account_id=account_id1,
                credit_account_id=account_id2,
                amount=Decimal("1000.00"),
            )

        # Check balances - should be unchanged.
        self.assertEqual(app.get_balance(account_id1), Decimal("50.00"))
        self.assertEqual(app.get_balance(account_id2), Decimal("100.00"))

        # Close account #1.
        app.close_account(account_id1)

        # Fail to transfer funds - account #1 is closed.
        with self.assertRaises(AccountClosedError):
            app.transfer_funds(
                debit_account_id=account_id1,
                credit_account_id=account_id2,
                amount=Decimal("50.00"),
            )

        # Fail to withdraw funds - account #1 is closed.
        with self.assertRaises(AccountClosedError):
            app.withdraw_funds(
                debit_account_id=account_id1,
                amount=Decimal("1.00"),
            )

        # Fail to deposit funds - account #1 is closed.
        with self.assertRaises(AccountClosedError):
            app.deposit_funds(
                credit_account_id=account_id1,
                amount=Decimal("1000.00"),
            )

        # Fail to set overdraft limit on account #1 - account is closed.
        with self.assertRaises(AccountClosedError):
            app.set_overdraft_limit(
                account_id=account_id1,
                overdraft_limit=Decimal("500.00"),
            )

        # Check balances - should be unchanged.
        self.assertEqual(app.get_balance(account_id1), Decimal("50.00"))
        self.assertEqual(app.get_balance(account_id2), Decimal("100.00"))

        # Check overdraft limits - should be unchanged.
        self.assertEqual(
            app.get_overdraft_limit(account_id1),
            Decimal("0.00"),
        )
        self.assertEqual(
            app.get_overdraft_limit(account_id2),
            Decimal("0.00"),
        )

        # Set overdraft limit on account #2.
        app.set_overdraft_limit(
            account_id=account_id2,
            overdraft_limit=Decimal("500.00"),
        )

        # Can't set negative overdraft limit.
        with self.assertRaises(AssertionError):
            app.set_overdraft_limit(
                account_id=account_id2,
                overdraft_limit=Decimal("-500.00"),
            )

        # Check overdraft limit of account #2.
        self.assertEqual(
            app.get_overdraft_limit(account_id2),
            Decimal("500.00"),
        )

        # Withdraw funds from account #2.
        app.withdraw_funds(
            debit_account_id=account_id2,
            amount=Decimal("500.00"),
        )

        # Check balance of account #2 - should be overdrawn.
        self.assertEqual(
            app.get_balance(account_id2),
            Decimal("-400.00"),
        )

        # Fail to withdraw funds from account #2 - insufficient funds.
        with self.assertRaises(InsufficientFundsError):
            app.withdraw_funds(
                debit_account_id=account_id2,
                amount=Decimal("101.00"),
            )