Eventing internals

Message envelope and metadata classes for the event transport layer. These classes define how domain events and commands are serialized, routed, and traced through the system.

MessageEnvelope

MessageEnvelope(*args: Any, **kwargs: Any)

Bases: BaseValueObject

Message envelope containing integrity and versioning information.

Source code in src/protean/core/value_object.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    # Support template dict pattern: VO({"key": "val"}, key2="val2")
    # Keyword args take precedence over template dict values.
    if args:
        merged: dict[str, Any] = {}
        for template in args:
            if not isinstance(template, dict):
                raise AssertionError(
                    f"Positional argument {template} passed must be a dict. "
                    f"This argument serves as a template for loading common "
                    f"values.",
                )
            merged.update(template)
        merged.update(kwargs)
        kwargs = merged
    try:
        super().__init__(**kwargs)
    except PydanticValidationError as e:
        raise ValidationError(convert_pydantic_errors(e))

compute_checksum `classmethod`

compute_checksum(payload: dict[str, Any]) -> str

Compute checksum for message integrity validation.

Source code in src/protean/utils/eventing.py

@classmethod
def compute_checksum(cls, payload: dict[str, Any]) -> str:
    """Compute checksum for message integrity validation."""
    json_data = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(json_data.encode("utf-8")).hexdigest()

Metadata

Metadata(*args: Any, **kwargs: Any)