orchestrata_ultimate.postman.payload

Reference › orchestrata_ultimate › postman › payload

Encoding, decoding, compression and validation of message bodies, SPEC-002 story FR-002 (REQ-002-FR2-001 to -010). PayloadHandler implements IPayloadHandler from interfaces and produces the MessagePayload value object defined in types. It is the body-side partner of envelope. No other module imports it.

create_payload(data, content_type, compress) encodes the data to bytes with encode, validates it and raises InvalidPayloadError when invalid. With compress=True it gzips the bytes, sets the content type to application/gzip and stores the content base64-encoded; other types are stored as UTF-8 text. encode handles each content type: JSON through _encode_json_value, text through _encode_text_value, and the binary and compressed types, which require bytes (_require_bytes_binary, _require_bytes_compressed). decode reverses it. validate rejects a payload larger than the maximum size (1 MB by default), an unsupported content type, JSON that does not parse and text that is not valid UTF-8. get_content_type picks a type from the data: JSON for dictionaries, lists, numbers, booleans and None, text for strings and UTF-8 bytes, binary for other bytes. compress and decompress use gzip.

At a glance

Class diagram

Dependency diagram

API

Classes

PayloadHandler

Bases: IPayloadHandler · Decorators: traces(spec='002', story='FR-002', reqs=['REQ-002-FR2-001', 'REQ-002-FR2-002', 'REQ-002-FR2-003', 'REQ-002-FR2-004', 'R… · Line 77

Handles message payload encoding, decoding, compression, and validation.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/postman/payload.py

# mypy: ignore-errors
"""
Payload handler for message encoding, decoding, compression, and validation.

This module implements the IPayloadHandler protocol for managing message payloads
in the Postman message broker, supporting multiple content types and compression.
"""

import base64
import gzip
import json
import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import (
    CONTENT_TYPE_BINARY,
    CONTENT_TYPE_COMPRESSED,
    CONTENT_TYPE_JSON,
    CONTENT_TYPE_TEXT,
    MAX_MESSAGE_SIZE_BYTES,
)
from orchestrata_ultimate.postman.exceptions import InvalidPayloadError
from orchestrata_ultimate.postman.interfaces import IPayloadHandler
from orchestrata_ultimate.postman.types import MessagePayload

# Spec: 002 - Postman Message Broker


logger = logging.getLogger(__name__)


@traces(spec="002", story="FR-002", reqs=["REQ-002-FR2-002"])
def _encode_json_value(data: Any) -> bytes:
    """Encode a non-bytes value as JSON bytes."""
    return json.dumps(data).encode("utf-8")


def _encode_text_value(data: Any) -> bytes:
    """Encode a non-bytes value as UTF-8 text bytes."""
    return str(data).encode("utf-8")


def _require_bytes_binary(data: Any) -> bytes:
    """Reject non-bytes data for the binary content type."""
    raise InvalidPayloadError("Binary content type requires bytes data")


def _require_bytes_compressed(data: Any) -> bytes:
    """Reject non-bytes data for the compressed content type."""
    raise InvalidPayloadError("Compressed content type requires bytes data")


# Encoders for non-bytes data, keyed by content type. Bytes inputs are passed
# through by ``encode`` before this table is consulted.
_NON_BYTES_ENCODERS = {
    CONTENT_TYPE_JSON: _encode_json_value,
    CONTENT_TYPE_TEXT: _encode_text_value,
    CONTENT_TYPE_BINARY: _require_bytes_binary,
    CONTENT_TYPE_COMPRESSED: _require_bytes_compressed,
}


@traces(
    spec="002",
    story="FR-002",
    reqs=[
        "REQ-002-FR2-001",
        "REQ-002-FR2-002",
        "REQ-002-FR2-003",
        "REQ-002-FR2-004",
        "REQ-002-FR2-006",
        "REQ-002-FR2-011",
        "REQ-002-FR2-012",
    ],
)
class PayloadHandler(IPayloadHandler):
    """
    Handles message payload encoding, decoding, compression, and validation.

    Supports multiple content types:
    - application/json: JSON payloads
    - text/plain: Plain text payloads
    - application/octet-stream: Binary payloads
    - application/gzip: Compressed payloads
    """

    @traces(
        spec="002",
        story="FR-002",
        reqs=[
            "REQ-002-FR2-006",
            "REQ-002-FR2-007",
            "REQ-002-FR2-008",
            "REQ-002-FR2-009",
            "REQ-002-FR2-010",
        ],
    )
    def validate(
        self, payload: bytes, content_type: str, max_size: int = MAX_MESSAGE_SIZE_BYTES
    ) -> tuple[bool, str | None]:
        """
        Validate payload size and content type.

        Args:
            payload: Payload bytes to validate
            content_type: MIME type of payload
            max_size: Maximum allowed size in bytes

        Returns:
            Tuple of (is_valid, error_message)

        Examples:
            >>> handler = PayloadHandler()
            >>> handler.validate(b"test", "application/json")
            (True, None)
            >>> handler.validate(b"x" * 2000000, "application/json")
            (False, "Payload exceeds maximum size")
        """
        # Check payload size
        if len(payload) > max_size:
            return (False, "Payload exceeds maximum size")

        # Validate content type is supported
        supported_types = {
            CONTENT_TYPE_JSON,
            CONTENT_TYPE_TEXT,
            CONTENT_TYPE_BINARY,
            CONTENT_TYPE_COMPRESSED,
        }

        if content_type not in supported_types:
            return (False, f"Unsupported content type: {content_type}")

        # For JSON content type, validate it is valid JSON
        if content_type == CONTENT_TYPE_JSON:
            try:
                json.loads(payload.decode("utf-8"))
            except (json.JSONDecodeError, UnicodeDecodeError) as e:
                return (False, f"Invalid JSON payload: {str(e)}")

        # For text content type, validate it is valid UTF-8
        if content_type == CONTENT_TYPE_TEXT:
            try:
                payload.decode("utf-8")
            except UnicodeDecodeError as e:
                return (False, f"Invalid UTF-8 text payload: {str(e)}")

        return (True, None)

    @traces(spec="002", story="FR-002", reqs=["REQ-002-FR2-004"])
    def compress(self, payload: bytes) -> bytes:
        """
        Compress payload using gzip compression.

        Args:
            payload: Payload bytes to compress

        Returns:
            Compressed payload bytes

        Raises:
            InvalidPayloadError: If compression fails

        Examples:
            >>> handler = PayloadHandler()
            >>> compressed = handler.compress(b"test data" * 100)
            >>> len(compressed) < len(b"test data" * 100)
            True
        """
        try:
            return gzip.compress(payload)
        except (OSError, TypeError, ValueError) as e:
            raise InvalidPayloadError(f"Compression failed: {str(e)}")

    @traces(spec="002", story="FR-002", reqs=["REQ-002-FR2-012"])
    def decompress(self, payload: bytes) -> bytes:
        """
        Decompress gzip-compressed payload.

        Args:
            payload: Compressed payload bytes

        Returns:
            Decompressed payload bytes

        Raises:
            InvalidPayloadError: If decompression fails

        Examples:
            >>> handler = PayloadHandler()
            >>> compressed = gzip.compress(b"test data")
            >>> handler.decompress(compressed)
            b'test data'
        """
        try:
            return gzip.decompress(payload)
        except (OSError, EOFError, TypeError, ValueError) as e:
            raise InvalidPayloadError(f"Decompression failed: {str(e)}")

    @traces(
        spec="002",
        story="FR-002",
        reqs=[
            "REQ-002-FR2-001",
            "REQ-002-FR2-002",
            "REQ-002-FR2-003",
            "REQ-002-FR2-004",
        ],
    )
    def create_payload(
        self,
        data: Any,
        content_type: str = CONTENT_TYPE_JSON,
        compress: bool = False,
    ) -> MessagePayload:
        """
        Create a MessagePayload from data with optional compression.

        Args:
            data: Data to encode (dict, str, bytes, etc.)
            content_type: MIME type for encoding
            compress: Whether to compress the payload

        Returns:
            MessagePayload instance

        Raises:
            InvalidPayloadError: If encoding or compression fails

        Examples:
            >>> handler = PayloadHandler()
            >>> payload = handler.create_payload({"key": "value"})
            >>> payload.content_type
            'application/json'
            >>> payload.compressed
            False
        """
        # Encode data to bytes
        payload_bytes = self.encode(data, content_type)

        # Validate payload
        is_valid, error = self.validate(payload_bytes, content_type)
        if not is_valid:
            # validate() always returns a non-None message alongside
            # is_valid=False (see its return sites); the fallback below
            # only guards the type, it's never reached in practice.
            raise InvalidPayloadError(error or "Payload validation failed")

        # Compress if requested
        if compress:
            payload_bytes = self.compress(payload_bytes)
            content_type = CONTENT_TYPE_COMPRESSED

        # Create MessagePayload
        # Convert bytes to string for MessagePayload.content
        # Compressed (gzip) bytes are not valid UTF-8, so base64-encode them
        if content_type == CONTENT_TYPE_COMPRESSED:
            content_str = base64.b64encode(payload_bytes).decode("ascii")
        elif isinstance(payload_bytes, bytes):
            content_str = payload_bytes.decode("utf-8")
        else:
            content_str = str(payload_bytes)
        return MessagePayload(
            content=content_str,
            content_type=content_type,
            compressed=compress,
        )

    @traces(
        spec="002",
        story="FR-002",
        reqs=[
            "REQ-002-FR2-001",
            "REQ-002-FR2-002",
            "REQ-002-FR2-003",
            "REQ-002-FR2-014",
        ],
    )
    def encode(self, data: Any, content_type: str) -> bytes:
        """
        Encode data to bytes based on content type.

        Args:
            data: Data to encode
            content_type: Target MIME type

        Returns:
            Encoded bytes

        Raises:
            InvalidPayloadError: If encoding fails

        Examples:
            >>> handler = PayloadHandler()
            >>> handler.encode({"key": "value"}, "application/json")
            b'{"key": "value"}'
        """
        try:
            encoder = _NON_BYTES_ENCODERS.get(content_type)
            if encoder is None:
                raise InvalidPayloadError(
                    f"Unsupported content type for encoding: {content_type}"
                )
            # All supported content types pass bytes through unchanged.
            if isinstance(data, bytes):
                return data
            return encoder(data)

        except InvalidPayloadError:
            raise
        except Exception as e:
            raise InvalidPayloadError(f"Encoding failed: {str(e)}")

    @traces(
        spec="002",
        story="FR-002",
        reqs=[
            "REQ-002-FR2-011",
            "REQ-002-FR2-012",
            "REQ-002-FR2-014",
        ],
    )
    def decode(self, payload: bytes, content_type: str) -> Any:
        """
        Decode payload bytes to Python objects based on content type.

        Args:
            payload: Payload bytes to decode
            content_type: MIME type for decoding

        Returns:
            Decoded Python object

        Raises:
            InvalidPayloadError: If decoding fails

        Examples:
            >>> handler = PayloadHandler()
            >>> handler.decode(b'{"key": "value"}', "application/json")
            {'key': 'value'}
        """
        try:
            if content_type == CONTENT_TYPE_JSON:
                return json.loads(payload.decode("utf-8"))

            elif content_type == CONTENT_TYPE_TEXT:
                return payload.decode("utf-8")

            elif content_type == CONTENT_TYPE_BINARY:
                return payload

            elif content_type == CONTENT_TYPE_COMPRESSED:
                return self.decompress(base64.b64decode(payload))

            else:
                raise InvalidPayloadError(
                    f"Unsupported content type for decoding: {content_type}"
                )

        except InvalidPayloadError:
            raise
        except Exception as e:
            raise InvalidPayloadError(f"Decoding failed: {str(e)}")

    def get_content_type(self, data: Any) -> str:
        """
        Auto-detect appropriate content type for data.

        Args:
            data: Data to analyze

        Returns:
            Appropriate MIME type string

        Examples:
            >>> handler = PayloadHandler()
            >>> handler.get_content_type({"key": "value"})
            'application/json'
            >>> handler.get_content_type("hello")
            'text/plain'
            >>> handler.get_content_type(b"\x00\x01\x02")
            'application/octet-stream'
        """
        if isinstance(data, (dict, list, int, float, bool)) or data is None:
            return CONTENT_TYPE_JSON

        elif isinstance(data, str):
            return CONTENT_TYPE_TEXT

        elif isinstance(data, bytes):
            # Check if bytes can be decoded as UTF-8 text
            try:
                data.decode("utf-8")
                return CONTENT_TYPE_TEXT
            except UnicodeDecodeError:
                return CONTENT_TYPE_BINARY

        else:
            # Default to JSON for other Python objects
            return CONTENT_TYPE_JSON