angzarr_client.destinations

Destinations context for sagas and process managers.

Provides type-safe access to destination sequences for command stamping. Sagas and PMs receive destination_sequences from the framework and use this class to stamp commands with the correct sequence numbers.

Design Philosophy:

Sagas and PMs should NOT rebuild destination state to make decisions. They receive only sequences for command stamping. Business logic belongs in aggregates - sagas/PMs just translate and coordinate.

Example usage:

@handles(OrderCreated) def handle_order(self, event: OrderCreated, destinations: Destinations):

cmd = ReserveInventory(order_id=event.order_id, quantity=event.quantity) destinations.stamp_command(cmd, “inventory”) return cmd

Classes

Destinations

Context for saga/PM handlers providing access to destination sequences.

Module Contents

class angzarr_client.destinations.Destinations(sequences: dict[str, int])

Context for saga/PM handlers providing access to destination sequences.

Sagas and PMs receive destination sequences (not full EventBooks) from the framework. This class provides a stamp_command() helper to set the correct sequence number on commands before they’re sent.

Why sequences only?

Sagas and PMs are translators/coordinators - they should NOT make business decisions based on destination state. If you need external information: 1. Inject it as a fact 2. Let aggregates decide (they validate commands) 3. Use sync mode for immediate feedback on command results

sequences

Map from domain name to next sequence number.

Create a Destinations context from a sequence map.

param sequences:

Dict mapping domain name to next sequence number. Typically from proto’s destination_sequences field.

classmethod from_proto(destination_sequences: dict[str, int]) → Destinations

Create from proto’s destination_sequences map.

Parameters:

destination_sequences – Map from ProcessManagerHandleRequest or SagaHandleRequest proto.

Returns:

Destinations instance with the sequences.

sequence_for(domain: str) → int | None

Get the next sequence number for a destination domain.

Parameters:

domain – The target domain name.

Returns:

The next sequence number, or None if domain not found.

stamp_command(cmd: angzarr_client.proto.angzarr.types_pb2.CommandBook, domain: str) → angzarr_client.proto.angzarr.types_pb2.CommandBook

Stamp a command with the correct sequence for its destination domain.

Modifies the command’s page headers in-place with the sequence number.

Parameters:

• cmd – The CommandBook to stamp.

• domain – The target domain (must be in destination_sequences).

Returns:

The same CommandBook (for chaining).

Raises:

InvalidArgumentError – If domain is not in destination_sequences. Carries code=MISSING_DESTINATION_SEQUENCE and details["domain"]=<domain> for cucumber assertions. Check your output_domains config when you see this. Audit #64.

static deferred_header(source_cover, source_seq: int) → angzarr_client.proto.angzarr.types_pb2.PageHeader

Build a PageHeader carrying an AngzarrDeferredSequence.

Accepts either a Cover wrapper (the canonical post-B2 form passed to handlers by the framework) or a raw Cover proto. Internally the proto is what gets copied into the AngzarrDeferredSequence payload.

Use this on saga-produced commands so the framework can dedupe on (source.root, source_seq, target.root). AMQP at-least-once redelivery of the trigger event then becomes a no-op at the destination aggregate’s pipeline (cached events returned without re-invoking business logic), instead of relying on a business guard that surfaces as an idempotent-failure-shaped retry storm.

has_domain(domain: str) → bool

Check if a destination domain is available.

Parameters:

domain – The domain name to check.

Returns:

True if the domain has a sequence entry.

property domains: list[str]

Get list of available destination domains.