Skip to main content

Aggregate Examples

Real handler examples from the poker domain. All code is from the actual examples/ directory.

⍼ Angzarr supports two implementation styles:

StyleDescriptionBest For
FunctionalPure guard()/validate()/compute() functionsSimple handlers, easy testing
OO (Object-Oriented)Aggregate class with @handles/[Handles] decoratorsRich domain models, encapsulation
LanguageFunctionalOO
Python✓✓
C#✓✓
Rust✓✓
Java✓✓
Go✓✓
C++✓✓

Command Handlers​

The examples below show the guard → validate → compute pattern (functional) and decorator-based dispatch (OO).

Pure functions following guard → validate → compute. Each function has a single responsibility, directly unit testable without mocks.

def deposit_guard(state: PlayerState) -> None:
    """Check state preconditions before processing deposit."""
    if not state.exists:
        raise CommandRejectedError("Player does not exist")


# docs:end:deposit_guard


# docs:start:deposit_validate
def deposit_validate(cmd: player.DepositFunds) -> int:
    """Validate deposit command and extract amount."""
    amount = cmd.amount.amount if cmd.amount else 0
    if amount <= 0:
        raise CommandRejectedError("amount must be positive")
    return amount


# docs:end:deposit_validate


# docs:start:deposit_compute
def deposit_compute(
    cmd: player.DepositFunds, state: PlayerState, amount: int
) -> player.FundsDeposited:
    """Build FundsDeposited event from validated inputs."""
    new_balance = state.bankroll + amount
    return player.FundsDeposited(
        amount=cmd.amount,
        new_balance=poker_types.Currency(amount=new_balance, currency_code="CHIPS"),
        deposited_at=now(),
    )


Two-Phase Reservation Pattern​

The Player aggregate demonstrates two-phase reservation. Funds are reserved before joining a table, then released if the join fails.

@command_handler(player.ReserveFunds)
def handle_reserve(
    cmd: player.ReserveFunds, state: PlayerState, seq: int
) -> player.FundsReserved:
    """Reserve funds for a table buy-in."""
    if not state.exists:
        raise CommandRejectedError("Player does not exist")

    amount = cmd.amount.amount if cmd.amount else 0
    if amount <= 0:
        raise CommandRejectedError("amount must be positive")

    table_key = cmd.table_root.hex()
    if table_key in state.table_reservations:
        raise CommandRejectedError("Funds already reserved for this table")

    if amount > state.available_balance:
        raise CommandRejectedError("Insufficient funds")

    new_reserved = state.reserved_funds + amount
    new_available = state.bankroll - new_reserved
    return player.FundsReserved(
        amount=cmd.amount,
        table_root=cmd.table_root,
        new_available_balance=poker_types.Currency(
            amount=new_available, currency_code="CHIPS"
        ),
        new_reserved_balance=poker_types.Currency(
            amount=new_reserved, currency_code="CHIPS"
        ),
        reserved_at=now(),
    )


State Building​

State is rebuilt by applying events. ⍼ Angzarr provides two patterns:

PatternDescriptionBest For
Functional (StateRouter)Fluent builder with typed event handlersClean separation, easy testing
OO with @appliesDecorators on aggregate class methodsRich domain models

StateRouter provides fluent registration of event appliers, keeping state reconstruction separate from command handling.

def apply_registered(state: PlayerState, event: player.PlayerRegistered) -> None:
    """Apply PlayerRegistered event to state."""
    state.player_id = f"player_{event.email}"
    state.display_name = event.display_name
    state.email = event.email
    state.player_type = event.player_type
    state.ai_model_id = event.ai_model_id
    state.status = "active"
    state.bankroll = 0
    state.reserved_funds = 0


def apply_deposited(state: PlayerState, event: player.FundsDeposited) -> None:
    """Apply FundsDeposited event to state."""
    if event.new_balance:
        state.bankroll = event.new_balance.amount


def apply_withdrawn(state: PlayerState, event: player.FundsWithdrawn) -> None:
    """Apply FundsWithdrawn event to state."""
    if event.new_balance:
        state.bankroll = event.new_balance.amount


def apply_reserved(state: PlayerState, event: player.FundsReserved) -> None:
    """Apply FundsReserved event to state."""
    if event.new_reserved_balance:
        state.reserved_funds = event.new_reserved_balance.amount
    table_key = event.table_root.hex()
    if event.amount:
        state.table_reservations[table_key] = event.amount.amount


def apply_released(state: PlayerState, event: player.FundsReleased) -> None:
    """Apply FundsReleased event to state."""
    if event.new_reserved_balance:
        state.reserved_funds = event.new_reserved_balance.amount
    table_key = event.table_root.hex()
    state.table_reservations.pop(table_key, None)


def apply_transferred(state: PlayerState, event: player.FundsTransferred) -> None:
    """Apply FundsTransferred event to state."""
    if event.new_balance:
        state.bankroll = event.new_balance.amount


Testing​

⍼ Angzarr uses Gherkin feature files as living specifications. The same feature files run against all language implementations, guaranteeing identical business behavior.

Example Gherkin feature:

Feature: Player fund reservation

  Players must reserve funds when joining a table to ensure
  they can cover their buy-in before sitting down.

  Scenario: Reserve funds for table buy-in
    Given a player with $500 available
    When the player reserves $200 for a table
    Then the player's available balance is $300
    And the player's reserved balance is $200

  Scenario: Cannot reserve more than available
    Given a player with $500 available
    When the player tries to reserve $600
    Then the request fails with "insufficient funds"
    And the player's available balance remains $500

The guard → validate → compute pattern makes handlers pure functions, directly testable without mocks:

def test_deposit_increases_bankroll():
    state = PlayerState(registered=True, bankroll=1000)
    cmd = DepositFunds(amount=500)

    event = compute_deposit(cmd, state)

    assert event.new_bankroll == 1500

See Testing for the full three-level testing strategy.

Next Steps​