Bank accounts application¶
This example demonstrates a straightforward event-sourced application.
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 not 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 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
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 decimal import Decimal
from uuid import UUID
from eventsourcing.application import AggregateNotFound, Application
from eventsourcing.examples.bankaccounts.domainmodel import BankAccount
class BankAccounts(Application[BankAccount]):
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
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.
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"),
)