angzarr_client.helpers¶

Pure utility helpers for working with Angzarr proto types.

Cover/book/page accessors live on the wrapper classes in angzarr_client.wrappers (e.g. Cover.domain(), EventBook.next_sequence()). What stays here are functions that either don’t take a wrapped proto, take a primitive, or build a new proto from scratch.

Attributes¶

`T`
`UNKNOWN_DOMAIN`
`WILDCARD_DOMAIN`
`DEFAULT_EDITION`
`META_ANGZARR_DOMAIN`
`PROJECTION_DOMAIN_PREFIX`
`PROJECTION_TYPE_URL`
`CORRELATION_ID_HEADER`
`TYPE_URL_PREFIX`
`type_url_matches_exact`
`full_type_url`

Functions¶

`uuid_to_proto`(→ angzarr_client.proto.angzarr.UUID)	Convert a Python UUID to a proto UUID.
`proto_to_uuid`(→ uuid.UUID)	Convert a proto UUID to Python UUID.
`proto_uuid_to_hex`(→ str)	Convert a proto UUID to hex string format.
`bytes_to_uuid_text`(→ str)	Convert raw bytes to a standard UUID text format.
`proto_uuid_to_text`(→ str)	Convert a proto UUID to its standard text format.
`implicit_edition`(→ angzarr_client.proto.angzarr.Edition)	Create an edition with the given name but no divergences.
`divergence_for`(→ int \| None)	Return the divergence sequence for a domain, or `None` if not found.
`events_from_response`(...)	Extract the event pages list from a `CommandResponse`.
`idempotency_key`(→ str \| None)	Build the composite idempotency key for a saga-produced deferred sequence.
`decode_event`(→ T \| None)	Decode a page's event payload if its type URL matches the given name.
`destination_map`(→ dict[str, ...)	Build a map from root UUID hex to EventBook for destination lookup.
`wire_name`(→ str)	Cross-language alias — identity in Python.
`type_url`(→ str)	Construct a full type URL from a fully-qualified message type name.
`type_name_from_url`(→ str)	Extract the wire-format type name from a type URL.
`type_url_matches`(→ bool)	Check if a type URL matches the given fully-qualified type name.
`type_matches`(→ bool)	Check if an Any contains a message of the given type using DESCRIPTOR.
`try_unpack`(→ T \| None)	Unpack an Any to msg_class if type matches, returning None otherwise.
`unpack`(→ T)	Unpack an Any to msg_class, raising ValueError if type doesn't match.
`full_type_name`(→ str)	Get the fully-qualified type name from a message class DESCRIPTOR.
`full_type_url_for`(→ str)	Get the full type URL for a message class.
`now`(→ google.protobuf.timestamp_pb2.Timestamp)	Return the current time as a protobuf Timestamp.
`parse_timestamp`(→ google.protobuf.timestamp_pb2.Timestamp)	Parse an RFC3339 timestamp string.
`new_cover`(→ angzarr_client.proto.angzarr.Cover)	Create a new Cover with the given parameters.
`new_command_page`(...)	Create a command page from a sequence and Any message.
`new_command_book`(...)	Create a CommandBook with the given cover and pages.
`range_selection`(...)	Create a sequence range selection.
`temporal_by_sequence`(...)	Create a temporal selection as-of a sequence.
`temporal_by_time`(...)	Create a temporal selection as-of a timestamp.
`correlated_metadata`(→ list[tuple[str, str]])	Build gRPC metadata carrying the `x-correlation-id` header.

Module Contents¶

angzarr_client.helpers.T¶

angzarr_client.helpers.UNKNOWN_DOMAIN = 'unknown'¶

angzarr_client.helpers.WILDCARD_DOMAIN = '*'¶

angzarr_client.helpers.DEFAULT_EDITION = ''¶

angzarr_client.helpers.META_ANGZARR_DOMAIN = '_angzarr'¶

angzarr_client.helpers.PROJECTION_DOMAIN_PREFIX = '_projection'¶

angzarr_client.helpers.PROJECTION_TYPE_URL = 'angzarr_client.proto.angzarr.Projection'¶

angzarr_client.helpers.CORRELATION_ID_HEADER = 'x-correlation-id'¶

angzarr_client.helpers.TYPE_URL_PREFIX = 'type.googleapis.com/'¶

angzarr_client.helpers.uuid_to_proto(u: uuid.UUID) → angzarr_client.proto.angzarr.UUID¶: Convert a Python UUID to a proto UUID.

angzarr_client.helpers.proto_to_uuid(u: angzarr_client.proto.angzarr.UUID) → uuid.UUID¶: Convert a proto UUID to Python UUID.

angzarr_client.helpers.proto_uuid_to_hex(u: angzarr_client.proto.angzarr.UUID | None) → str¶: Convert a proto UUID to hex string format.

angzarr_client.helpers.bytes_to_uuid_text(b: bytes) → str¶

Convert raw bytes to a standard UUID text format.

16-byte input formats as the canonical 8-4-4-4-12 UUID; other lengths fall back to hex.

angzarr_client.helpers.proto_uuid_to_text(u: angzarr_client.proto.angzarr.UUID | None) → str¶: Convert a proto UUID to its standard text format.

angzarr_client.helpers.implicit_edition(name: str) → angzarr_client.proto.angzarr.Edition¶: Create an edition with the given name but no divergences.

angzarr_client.helpers.divergence_for(e: angzarr_client.proto.angzarr.Edition | None, domain_name: str) → int | None¶: Return the divergence sequence for a domain, or None if not found.

angzarr_client.helpers.events_from_response(resp) → list[angzarr_client.proto.angzarr.EventPage]¶

Extract the event pages list from a CommandResponse.

Returns [] for a missing/empty events book. The wrapper-method equivalent is CommandResponse(resp).events() (which returns a list of wrapped pages); this free function returns raw protos for the common case where the caller just wants to iterate them.

angzarr_client.helpers.idempotency_key(deferred) → str | None¶

Build the composite idempotency key for a saga-produced deferred sequence.

Format: {source.edition}:{source.domain}:{source.root_hex}:{source_seq}. Returns None when the deferred sequence has no source cover — a malformed wire input is a missing-key signal, not an exception. Audit finding #55.

angzarr_client.helpers.decode_event(page: angzarr_client.proto.angzarr.EventPage, full_type_name: str, msg_class: type[T]) → T | None¶

Decode a page’s event payload if its type URL matches the given name.

Runtime-name dispatch — full_type_name is the fully-qualified proto name (e.g. "orders.OrderCreated"). Compared exactly against TYPE_URL_PREFIX + full_type_name. The class-keyed shortcut is EventPage(page).decode_typed(msg_class) (derives the name from msg_class.DESCRIPTOR.full_name); use this free function when the type name is a runtime variable.

angzarr_client.helpers.destination_map(destinations: list[angzarr_client.proto.angzarr.EventBook]) → dict[str, angzarr_client.proto.angzarr.EventBook]¶

Build a map from root UUID hex to EventBook for destination lookup.

Used in multi-destination sagas to look up the correct EventBook by aggregate root when setting command sequences. Entries without a root are skipped.

angzarr_client.helpers.wire_name(full_name: str) → str¶

Cross-language alias — identity in Python.

Audit finding #32. Python’s protoc emits DESCRIPTOR.full_name and Pack() type URLs with the full angzarr_client.proto.* package prefix already, so no stripping is needed. The shim exists so cross-language code can call wire_name in either language.

angzarr_client.helpers.type_url(type_name: str) → str¶: Construct a full type URL from a fully-qualified message type name.

angzarr_client.helpers.type_name_from_url(type_url_str: str) → str¶

Extract the wire-format type name from a type URL.

Returns the part after the last /. Inputs with no slash are returned unchanged.

angzarr_client.helpers.type_url_matches(type_url_str: str, type_name: str) → bool¶: Check if a type URL matches the given fully-qualified type name.

angzarr_client.helpers.type_url_matches_exact¶

angzarr_client.helpers.type_matches(any_proto: google.protobuf.any_pb2.Any, msg_class: type[T]) → bool¶: Check if an Any contains a message of the given type using DESCRIPTOR.

angzarr_client.helpers.try_unpack(any_proto: google.protobuf.any_pb2.Any, msg_class: type[T]) → T | None¶: Unpack an Any to msg_class if type matches, returning None otherwise.

angzarr_client.helpers.unpack(any_proto: google.protobuf.any_pb2.Any, msg_class: type[T]) → T¶: Unpack an Any to msg_class, raising ValueError if type doesn’t match.

angzarr_client.helpers.full_type_name(msg_class: type[google.protobuf.message.Message]) → str¶: Get the fully-qualified type name from a message class DESCRIPTOR.

angzarr_client.helpers.full_type_url_for(msg_class: type[google.protobuf.message.Message]) → str¶: Get the full type URL for a message class.

angzarr_client.helpers.full_type_url¶

angzarr_client.helpers.now() → google.protobuf.timestamp_pb2.Timestamp¶: Return the current time as a protobuf Timestamp.

angzarr_client.helpers.parse_timestamp(rfc3339: str) → google.protobuf.timestamp_pb2.Timestamp¶: Parse an RFC3339 timestamp string.

angzarr_client.helpers.new_cover(domain_name: str, root: uuid.UUID, correlation_id_val: str = '', edition_val: angzarr_client.proto.angzarr.Edition | None = None) → angzarr_client.proto.angzarr.Cover¶: Create a new Cover with the given parameters.

angzarr_client.helpers.new_command_page(sequence: int, command: google.protobuf.any_pb2.Any) → angzarr_client.proto.angzarr.CommandPage¶: Create a command page from a sequence and Any message.

angzarr_client.helpers.new_command_book(cover: angzarr_client.proto.angzarr.Cover, pages: list[angzarr_client.proto.angzarr.CommandPage]) → angzarr_client.proto.angzarr.CommandBook¶: Create a CommandBook with the given cover and pages.

angzarr_client.helpers.range_selection(lower: int, upper: int | None = None) → angzarr_client.proto.angzarr.SequenceRange¶: Create a sequence range selection.

angzarr_client.helpers.temporal_by_sequence(seq: int) → angzarr_client.proto.angzarr.TemporalQuery¶: Create a temporal selection as-of a sequence.

angzarr_client.helpers.temporal_by_time(ts: google.protobuf.timestamp_pb2.Timestamp) → angzarr_client.proto.angzarr.TemporalQuery¶: Create a temporal selection as-of a timestamp.

angzarr_client.helpers.correlated_metadata(correlation_id: str) → list[tuple[str, str]]¶

Build gRPC metadata carrying the x-correlation-id header.

Returns the list shape grpcio expects on stub calls. Empty correlation IDs produce an empty list — the header is skipped, never sent as a blank value. Audit #69.