orchestrata_ultimate.cxp.payload

Reference › orchestrata_ultimate › cxp › payload

PayloadManager reads, writes and checks the payload of a CXPPacket (Spec 004 US-003). JSON payloads (dicts, lists and scalars) are stored on the packet as they are. Binary payloads are stored in a transport envelope, base64-encoded and optionally gzip-compressed, so the packet stays JSON-safe. set_payload chooses the path from the content type (JSON by default, anything else is binary), accepts None, and raises InvalidPacketError if the data type does not match the content type or the size limit is exceeded. get_payload returns the JSON value or decodes an envelope back to bytes, and get_payload_json and get_payload_binary return the string or bytes form. is_json, is_binary, get_content_type and get_payload_size describe the stored payload, and validate_payload_size checks it against separate limits for JSON (10 MB) and binary (100 MB) payloads. Packets are serialized by json_serializer. See cxp and exceptions.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PayloadManager

Decorators: traces(spec='004', story='US-003', reqs=['REQ-004-US-003-001', 'REQ-004-US-003-008', 'REQ-004-US-003-018', 'REQ-004-US-… · Line 62

Manages CXP packet payloads: access, type detection, validation.

Source

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

# mypy: ignore-errors
"""
CXP Payload Manager.

Handles payload access, content-type detection, size validation,
and JSON/binary serialization for CXP packets.

Spec: 004-cxp, Section US-003
Author: Mohamed

Payload Limits:
    - JSON payloads: max 10 MB (10_485_760 bytes)
    - Binary payloads: max 100 MB (104_857_600 bytes)

Binary payloads are base64-encoded for transport since CXP packets
serialize to JSON. The PayloadManager transparently handles encoding
and decoding.
"""

# Spec: 004 - CXP Protocol

import base64
import gzip
import json
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.cxp.exceptions import InvalidPacketError

__all__ = [
    "PayloadManager",
    "JSON_MAX_BYTES",
    "BINARY_MAX_BYTES",
]


# ============================================================================
# Constants
# ============================================================================

JSON_MAX_BYTES: int = 10_485_760  # 10 MB
BINARY_MAX_BYTES: int = 104_857_600  # 100 MB


# ============================================================================
# PayloadManager
# ============================================================================


@traces(
    spec="004",
    story="US-003",
    reqs=[
        "REQ-004-US-003-001",
        "REQ-004-US-003-008",
        "REQ-004-US-003-018",
        "REQ-004-US-003-020",
        "REQ-004-US-003-024",
    ],
)
class PayloadManager:
    """Manages CXP packet payloads: access, type detection, validation.

    Provides a clean interface for reading/writing payloads on CXPPacket
    instances, with content-type detection and size enforcement per
    Spec 004 Section US-003.

    Usage::

        pm = PayloadManager()

        # Set a JSON payload
        packet = CXPPacket(from_agent="U:Ox", to_agent="U:Cd", subject="task")
        pm.set_payload(packet, {"action": "build", "target": "api"})

        # Read it back
        data = pm.get_payload(packet)       # -> dict
        raw = pm.get_payload_json(packet)    # -> JSON string

        # Binary payload
        pm.set_payload(packet, b"\x89PNG...", content_type="application/octet-stream")
        raw_bytes = pm.get_payload_binary(packet)  # -> bytes
    """

    # ------------------------------------------------------------------
    # Read methods
    # ------------------------------------------------------------------

    @traces(spec="004", story="US-003", reqs=["REQ-004-US-003-020"])
    def get_payload(self, packet: CXPPacket) -> Any:
        """Return the payload from a packet.

        For JSON payloads, returns the dict/list/scalar directly.
        For binary payloads stored as base64 strings (transport form),
        decodes and returns the raw bytes.

        Args:
            packet: The CXP packet to read from.

        Returns:
            The payload value. Type depends on what was stored:
            dict/list for JSON, bytes for binary, or None if empty.
        """
        payload = packet.payload
        if payload is None:
            return None

        # If payload is already bytes, return as-is
        if isinstance(payload, (bytes, bytearray)):
            return bytes(payload)

        # If payload is a dict with our binary transport envelope,
        # decode the base64 content back to bytes
        if isinstance(payload, dict) and payload.get("_binary_transport"):
            encoded = payload.get("data", "")
            compressed = payload.get("compressed", False)
            raw = base64.b64decode(encoded)
            if compressed:
                raw = gzip.decompress(raw)
            return raw

        return payload

    @traces(
        spec="004",
        story="US-003",
        reqs=["REQ-004-US-003-021", "REQ-004-US-003-023"],
    )
    def get_payload_json(self, packet: CXPPacket) -> str:
        """Return the payload as a JSON string.

        Args:
            packet: The CXP packet to read from.

        Returns:
            JSON string representation of the payload.

        Raises:
            InvalidPacketError: If the payload cannot be serialized to JSON
                (e.g. raw bytes without the transport envelope).
        """
        payload = packet.payload
        if payload is None:
            return "null"

        # Binary transport envelope is JSON-serializable as-is
        if isinstance(payload, (bytes, bytearray)):
            raise InvalidPacketError(
                "Cannot serialize raw binary payload to JSON. "
                "Use set_payload() with content_type='application/octet-stream' "
                "to wrap binary data in a transport envelope.",
                field="payload",
            )

        try:
            return json.dumps(payload, default=str, ensure_ascii=False)
        except (TypeError, ValueError) as exc:
            raise InvalidPacketError(
                f"Payload is not JSON-serializable: {exc}",
                field="payload",
            ) from exc

    @traces(
        spec="004",
        story="US-003",
        reqs=["REQ-004-US-003-022"],
    )
    def get_payload_binary(self, packet: CXPPacket) -> bytes:
        """Return the payload as bytes.

        For binary payloads, returns the raw bytes (decoding base64
        transport envelope if necessary).
        For JSON payloads, returns the UTF-8 encoded JSON string.

        Args:
            packet: The CXP packet to read from.

        Returns:
            Bytes representation of the payload.
        """
        payload = packet.payload
        if payload is None:
            return b""

        # Already raw bytes
        if isinstance(payload, (bytes, bytearray)):
            return bytes(payload)

        # Binary transport envelope -> decode
        if isinstance(payload, dict) and payload.get("_binary_transport"):
            encoded = payload.get("data", "")
            compressed = payload.get("compressed", False)
            raw = base64.b64decode(encoded)
            if compressed:
                raw = gzip.decompress(raw)
            return raw

        # JSON payload -> encode to UTF-8 bytes
        json_str = json.dumps(payload, default=str, ensure_ascii=False)
        return json_str.encode("utf-8")

    # ------------------------------------------------------------------
    # Write methods
    # ------------------------------------------------------------------

    @traces(
        spec="004",
        story="US-003",
        reqs=["REQ-004-US-003-002", "REQ-004-US-003-003"],
    )
    def set_payload(
        self,
        packet: CXPPacket,
        data: Any,
        content_type: str = "application/json",
        compress: bool = False,
    ) -> None:
        """Set the payload on a packet with validation.

        For JSON content types, the data is stored directly (dict, list, etc.).
        For binary content types, the data (bytes) is base64-encoded and
        wrapped in a transport envelope for JSON-safe serialization.

        Args:
            packet: The CXP packet to modify.
            data: The payload data. Must be bytes for binary content types,
                  or a JSON-serializable value for JSON content types.
            content_type: MIME type. Defaults to "application/json".
                Use "application/octet-stream" or any non-JSON type for binary.
            compress: If True, gzip-compress binary payloads before base64
                encoding. Ignored for JSON payloads.

        Raises:
            InvalidPacketError: If the payload exceeds size limits or
                the data type doesn't match the content type.
        """
        if data is None:
            packet.payload = None
            return

        is_binary = not self._is_json_content_type(content_type)

        if is_binary:
            self._set_binary_payload(packet, data, content_type, compress)
        else:
            self._set_json_payload(packet, data)

    # ------------------------------------------------------------------
    # Type detection
    # ------------------------------------------------------------------

    @traces(
        spec="004",
        story="US-003",
        reqs=["REQ-004-US-003-024"],
    )
    def is_json(self, packet: CXPPacket) -> bool:
        """Return True if the payload is a JSON-type value (dict, list, scalar).

        Binary transport envelopes are NOT considered JSON payloads even
        though they are technically dicts.

        Args:
            packet: The CXP packet to inspect.

        Returns:
            True if the payload is a dict (non-binary-envelope), list,
            str, int, float, or bool. False otherwise.
        """
        payload = packet.payload
        if payload is None:
            return False

        if isinstance(payload, (bytes, bytearray)):
            return False

        if isinstance(payload, dict) and payload.get("_binary_transport"):
            return False

        return isinstance(payload, (dict, list, str, int, float, bool))

    @traces(
        spec="004",
        story="US-003",
        reqs=["REQ-004-US-003-024"],
    )
    def is_binary(self, packet: CXPPacket) -> bool:
        """Return True if the payload is binary data.

        This includes both raw bytes and base64 transport envelopes.

        Args:
            packet: The CXP packet to inspect.

        Returns:
            True if the payload is bytes, bytearray, or a binary
            transport envelope dict.
        """
        payload = packet.payload
        if payload is None:
            return False

        if isinstance(payload, (bytes, bytearray)):
            return True

        if isinstance(payload, dict) and payload.get("_binary_transport"):
            return True

        return False

    @traces(spec="004", story="US-003", reqs=["REQ-004-US-003-012"])
    def get_content_type(self, packet: CXPPacket) -> str:
        """Detect and return the content type of the payload.

        Args:
            packet: The CXP packet to inspect.

        Returns:
            MIME type string. Returns "application/json" for JSON payloads,
            the stored content type for binary transport envelopes, or
            "application/octet-stream" for raw bytes.
            Returns "application/x-empty" for None payloads.
        """
        payload = packet.payload
        if payload is None:
            return "application/x-empty"

        if isinstance(payload, (bytes, bytearray)):
            return "application/octet-stream"

        if isinstance(payload, dict) and payload.get("_binary_transport"):
            return payload.get("content_type", "application/octet-stream")

        return "application/json"

    # ------------------------------------------------------------------
    # Size methods
    # ------------------------------------------------------------------

    @traces(
        spec="004", story="US-003", reqs=["REQ-004-US-003-007", "REQ-004-US-003-011"]
    )
    def get_payload_size(self, packet: CXPPacket) -> int:
        """Return the size of the payload in bytes.

        For JSON payloads, this is the byte length of the UTF-8 encoded
        JSON string. For binary payloads, this is the raw byte count
        (before base64 encoding, after decompression if applicable).

        Args:
            packet: The CXP packet to measure.

        Returns:
            Size in bytes, or 0 for None payloads.
        """
        payload = packet.payload
        if payload is None:
            return 0

        if isinstance(payload, (bytes, bytearray)):
            return len(payload)

        if isinstance(payload, dict) and payload.get("_binary_transport"):
            # Report the decoded size, not the base64 size
            encoded = payload.get("data", "")
            compressed = payload.get("compressed", False)
            raw = base64.b64decode(encoded)
            if compressed:
                raw = gzip.decompress(raw)
            return len(raw)

        # JSON payload: measure the serialized form
        json_str = json.dumps(payload, default=str, ensure_ascii=False)
        return len(json_str.encode("utf-8"))

    @traces(
        spec="004",
        story="US-001",
        reqs=[
            "REQ-004-US-003-007",
            "REQ-004-US-003-011",
            "REQ-004-US-003-018",
        ],
    )
    def validate_payload_size(self, packet: CXPPacket) -> bool:
        """Check whether the payload is within size limits.

        JSON payloads must be <= JSON_MAX_BYTES (10 MB).
        Binary payloads must be <= BINARY_MAX_BYTES (100 MB).

        Args:
            packet: The CXP packet to validate.

        Returns:
            True if the payload is within the allowed size limit,
            False otherwise. Returns True for None payloads.
        """
        payload = packet.payload
        if payload is None:
            return True

        size = self.get_payload_size(packet)

        if self.is_binary(packet):
            return size <= BINARY_MAX_BYTES

        return size <= JSON_MAX_BYTES

    # ------------------------------------------------------------------
    # Private helpers
    # ------------------------------------------------------------------

    @traces(spec="004", story="US-003", reqs=["REQ-004-US-003-016"])
    def _set_json_payload(self, packet: CXPPacket, data: Any) -> None:
        """Validate and store a JSON payload on the packet.

        Args:
            packet: Target packet.
            data: JSON-serializable value.

        Raises:
            InvalidPacketError: If data is bytes or exceeds JSON_MAX_BYTES.
        """
        if isinstance(data, (bytes, bytearray)):
            raise InvalidPacketError(
                "Cannot store bytes as JSON payload. "
                "Use content_type='application/octet-stream' for binary data.",
                field="payload",
            )

        # Verify it is serializable and measure size
        try:
            json_str = json.dumps(data, default=str, ensure_ascii=False)
        except (TypeError, ValueError) as exc:
            raise InvalidPacketError(
                f"Payload is not JSON-serializable: {exc}",
                field="payload",
            ) from exc

        size = len(json_str.encode("utf-8"))
        if size > JSON_MAX_BYTES:
            raise InvalidPacketError(
                f"JSON payload size {size:,} bytes exceeds limit of "
                f"{JSON_MAX_BYTES:,} bytes (10 MB).",
                field="payload",
                details={"size": size, "limit": JSON_MAX_BYTES},
            )

        packet.payload = data

    @traces(
        spec="004",
        story="US-003",
        reqs=[
            "REQ-004-US-003-008",
            "REQ-004-US-003-013",
            "REQ-004-US-003-017",
        ],
    )
    def _set_binary_payload(
        self,
        packet: CXPPacket,
        data: Any,
        content_type: str,
        compress: bool,
    ) -> None:
        """Validate and store a binary payload as a transport envelope.

        Binary data is base64-encoded and wrapped in a dict envelope so
        the overall CXP packet remains JSON-serializable.

        Args:
            packet: Target packet.
            data: Must be bytes or bytearray.
            content_type: MIME type to store in the envelope.
            compress: If True, gzip-compress before base64 encoding.

        Raises:
            InvalidPacketError: If data is not bytes or exceeds BINARY_MAX_BYTES.
        """
        if not isinstance(data, (bytes, bytearray)):
            raise InvalidPacketError(
                f"Binary content type '{content_type}' requires bytes data, "
                f"got {type(data).__name__}.",
                field="payload",
            )

        raw = bytes(data)

        if len(raw) > BINARY_MAX_BYTES:
            raise InvalidPacketError(
                f"Binary payload size {len(raw):,} bytes exceeds limit of "
                f"{BINARY_MAX_BYTES:,} bytes (100 MB).",
                field="payload",
                details={"size": len(raw), "limit": BINARY_MAX_BYTES},
            )

        if compress:
            transport_bytes = gzip.compress(raw)
        else:
            transport_bytes = raw

        encoded = base64.b64encode(transport_bytes).decode("ascii")

        packet.payload = {
            "_binary_transport": True,
            "content_type": content_type,
            "data": encoded,
            "compressed": compress,
            "original_size": len(raw),
        }

    @staticmethod
    @traces(spec="004", story="US-003", reqs=["REQ-004-US-003-012"])
    def _is_json_content_type(content_type: str) -> bool:
        """Return True if the content type indicates JSON data.

        Args:
            content_type: MIME type string.

        Returns:
            True for 'application/json' and variants like
            'application/vnd.api+json'.
        """
        ct = content_type.strip().lower()
        if ct == "application/json":
            return True
        if ct.startswith("application/") and ct.endswith("+json"):
            return True
        if ct == "text/json":
            return True
        return False