Source code for examples.bankaccounts.domainmodel

from __future__ import annotations

from decimal import Decimal

from eventsourcing.domain import Aggregate, event


[docs] 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
[docs] @event("Credited") def credit(self, amount: Decimal) -> None: self.check_account_is_not_closed() self.balance += amount
[docs] @event("Debited") def debit(self, amount: Decimal) -> None: self.check_account_is_not_closed() self.check_has_sufficient_funds(amount) self.balance -= amount
[docs] @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
[docs] @event("Closed") def close(self) -> None: self.is_closed = True
[docs] def check_account_is_not_closed(self) -> None: if self.is_closed: raise AccountClosedError({"account_id": self.id})
[docs] def check_has_sufficient_funds(self, amount: Decimal) -> None: if self.balance - amount < -self.overdraft_limit: raise InsufficientFundsError({"account_id": self.id})
[docs] class TransactionError(Exception): pass
[docs] class AccountClosedError(TransactionError): pass
[docs] class InsufficientFundsError(TransactionError): pass