Banking Domain Example

thông tin

This example demonstrates a banking transaction domain with deposits, withdrawals, and transfers.

Overview

This example shows a complete banking domain with:

• Account management
• Transaction processing
• Money transfers
• Balance tracking

Complete Example

Aggregate Definition

from fluvius.domain import Aggregate, action
from fluvius.data import DataModel, field, Decimal

class AccountState(DataModel):
    balance: Decimal = field(initial=Decimal('0.00'))
    account_number: str = field()
    owner_name: str = field()

class TransactionAggregate(Aggregate):
    def __init__(self, domain):
        super().__init__(domain)
        self._state = None
    
    @action(evt_key='account-opened', resources=['bank-account'])
    async def open_account(self, account_number: str, owner_name: str, initial_deposit: Decimal):
        """Open a new bank account"""
        if self._state is not None:
            raise ValueError("Account already exists")
        
        self._state = AccountState(
            account_number=account_number,
            owner_name=owner_name,
            balance=initial_deposit
        )
        return self._state
    
    @action(evt_key='money-deposited', resources=['bank-account'])
    async def deposit_money(self, amount: Decimal):
        """Deposit money into account"""
        if self._state is None:
            raise ValueError("Account does not exist")
        if amount <= 0:
            raise ValueError("Deposit amount must be positive")
        
        self._state = self._state.update(
            balance=self._state.balance + amount
        )
        return self._state
    
    @action(evt_key='money-withdrawn', resources=['bank-account'])
    async def withdraw_money(self, amount: Decimal):
        """Withdraw money from account"""
        if self._state is None:
            raise ValueError("Account does not exist")
        if amount <= 0:
            raise ValueError("Withdrawal amount must be positive")
        if self._state.balance < amount:
            raise ValueError("Insufficient funds")
        
        self._state = self._state.update(
            balance=self._state.balance - amount
        )
        return self._state
    
    @action(evt_key='money-transferred', resources=['bank-account'])
    async def transfer_money(self, amount: Decimal, recipient_id: UUID_TYPE):
        """Transfer money to another account"""
        if self._state is None:
            raise ValueError("Account does not exist")
        if amount <= 0:
            raise ValueError("Transfer amount must be positive")
        if self._state.balance < amount:
            raise ValueError("Insufficient funds")
        
        # Withdraw from this account
        self._state = self._state.update(
            balance=self._state.balance - amount
        )
        
        # Deposit to recipient (would be handled by recipient's aggregate)
        # This is a simplified example
        return self._state

Domain Definition

from fluvius.domain import Domain
from .aggregate import TransactionAggregate

class TransactionManagerDomain(Domain):
    __aggregate__ = TransactionAggregate
    
    class Meta:
        revision = 1
        tags = ["banking", "transactions", "finance"]
        title = "Banking Transaction Domain"
        description = "Domain for managing bank account transactions"

Usage

import asyncio
from decimal import Decimal
from fluvius.domain.context import SanicContext
from fluvius.data import UUID_GENR
from banking_domain import TransactionManagerDomain

async def main():
    ctx = SanicContext.create(namespace='bank-account')
    domain = TransactionManagerDomain(ctx)
    
    account_id = UUID_GENR()
    domain.set_aggroot('bank-account', account_id)
    
    # Open account
    await domain.process_command(
        domain.create_command('open-account', {
            'account_number': '123456789',
            'owner_name': 'John Doe',
            'initial_deposit': Decimal('1000.00')
        })
    )
    
    # Check balance
    account = await domain.statemgr.fetch('bank-account', account_id)
    print(f'Initial balance: {account.balance}')
    
    # Process transactions
    commands = [
        domain.create_command('withdraw-money', {'amount': Decimal('20.00')}),
        domain.create_command('deposit-money', {'amount': Decimal('10.00')}),
        domain.create_command('transfer-money', {
            'amount': Decimal('30.00'),
            'recipient': UUID_GENR()
        })
    ]
    
    async for response in domain.command_processor.process(*commands):
        print(f'Transaction: {response}')
    
    # Final balance
    account = await domain.statemgr.fetch('bank-account', account_id)
    print(f'Final balance: {account.balance}')

if __name__ == '__main__':
    asyncio.run(main())

Key Features

1. Business Rules: Enforced in aggregate methods
2. Validation: Input validation before state changes
3. Transactions: Multiple operations in sequence
4. State Queries: Check balance at any time
5. Event Generation: All actions generate events automatically

Next Steps

• Learn about State Management
• Check out FastAPI Integration
• Read the Domain API Reference