orchestrata_ultimate.cxp.json_serializer

Reference › orchestrata_ultimate › cxp › json_serializer

JSONSerializer is the JSON form of the ISerializer contract from interfaces (Spec 004 US-006). to_json turns a CXPPacket into a string, with indented output on request, and from_json rebuilds the packet; to_dict and from_dict are the dictionary steps. Plain data goes through json.dumps directly; when the dictionary holds values JSON cannot represent, _needs_custom_encoding selects CXPEncoder. That encoder writes datetimes as ISO 8601, enums as their value, bytes as a marked base64 object, sets as sorted lists, and replaces NaN and infinity with null; _decode_bytes_hook restores the bytes on the way back. Failures raise SerializationError or DeserializationError from exceptions, tagged with the format and packet id. The compact alternative is binary_serializer. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPEncoder

Bases: json.JSONEncoder · Decorators: traces(spec='004', story='US-003', reqs=['REQ-004-US-003-009', 'REQ-004-US-006-003', 'REQ-004-US-006-004', 'REQ-004-US-… · Line 44

Custom JSON encoder handling CXP-specific types.

JSONSerializer

Decorators: traces(spec='004', story='US-006', reqs=['REQ-004-US-006-001', 'REQ-004-US-006-009']) · Line 113

JSON serializer for CXP packets.

Functions

Source

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

# mypy: ignore-errors
"""
CXP JSON Serializer.

JSON serialization and deserialization for CXP packets.
Implements ISerializer JSON methods per Spec 004 Section US-006.

Performance targets:
    - to_json: < 10ms for 1KB payload
    - from_json: < 10ms

Author: Mohamed
"""

# Spec: 004 - CXP Protocol

import base64
import json
import logging
import math
from datetime import datetime
from enum import Enum
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 DeserializationError, SerializationError

__all__ = ["JSONSerializer"]

logger = logging.getLogger(__name__)


@traces(
    spec="004",
    story="US-003",
    reqs=[
        "REQ-004-US-003-009",
        "REQ-004-US-006-003",
        "REQ-004-US-006-004",
        "REQ-004-US-006-006",
    ],
)
class CXPEncoder(json.JSONEncoder):
    """Custom JSON encoder handling CXP-specific types.

    Handles:
        - datetime  -> ISO 8601 string
        - Enum      -> .value
        - bytes     -> base64-encoded string with marker
        - NaN/Inf   -> null (JSON has no NaN/Infinity)
        - sets      -> sorted list
    """

    @traces(
        spec="004",
        story="US-003",
        reqs=[
            "REQ-004-US-003-009",
            "REQ-004-US-006-003",
            "REQ-004-US-006-004",
            "REQ-004-US-006-006",
        ],
    )
    def default(self, o: Any) -> Any:
        if isinstance(o, datetime):
            return o.isoformat()
        if isinstance(o, Enum):
            return o.value
        if isinstance(o, bytes):
            return {"__bytes__": base64.b64encode(o).decode("ascii")}
        if isinstance(o, set):
            return sorted(o, key=str)
        if isinstance(o, frozenset):
            return sorted(o, key=str)
        return super().default(o)

    def encode(self, o: Any) -> str:
        """Override encode to sanitize NaN/Infinity before serialization."""
        sanitized = self._sanitize(o)
        return super().encode(sanitized)

    def _sanitize(self, obj: Any) -> Any:
        """Replace NaN and Infinity with None (JSON-safe)."""
        if isinstance(obj, float):
            if math.isnan(obj) or math.isinf(obj):
                return None
            return obj
        if isinstance(obj, dict):
            return {k: self._sanitize(v) for k, v in obj.items()}
        if isinstance(obj, (list, tuple)):
            return [self._sanitize(item) for item in obj]
        return obj


@traces(
    spec="004",
    story="US-003",
    reqs=["REQ-004-US-003-010", "REQ-004-US-006-014"],
)
def _decode_bytes_hook(obj: dict[str, Any]) -> Any:
    """JSON object_hook that restores base64-encoded bytes."""
    if "__bytes__" in obj and len(obj) == 1:
        return base64.b64decode(obj["__bytes__"])
    return obj


@traces(
    spec="004",
    story="US-006",
    reqs=["REQ-004-US-006-001", "REQ-004-US-006-009"],
)
class JSONSerializer:
    """JSON serializer for CXP packets.

    Conforms to ISerializer JSON methods (Spec 004 Section US-006).

    Usage:
        serializer = JSONSerializer()
        json_str = serializer.to_json(packet)
        restored = serializer.from_json(json_str)
    """

    def __init__(self) -> None:
        self._encoder = CXPEncoder
        logger.debug("JSONSerializer initialized")

    @staticmethod
    @traces(spec="004", story="US-006", reqs=["REQ-004-PERF-001"])
    def _value_needs_encoding(value: Any) -> bool:
        """Return True if *value* itself is a type CXPEncoder must handle.

        Covers datetime, Enum, bytes, set, frozenset, and non-finite floats
        (NaN/Infinity).  Container recursion is handled by the caller.
        """
        if isinstance(value, (datetime, Enum, bytes, set, frozenset)):
            return True
        return isinstance(value, float) and (math.isnan(value) or math.isinf(value))

    @staticmethod
    @traces(spec="004", story="US-006", reqs=["REQ-004-PERF-001"])
    def _needs_custom_encoding(data: dict) -> bool:
        """Check if data contains types requiring custom encoding.

        Returns True if any value is datetime, Enum, bytes, set, frozenset,
        NaN, or Infinity — types that CXPEncoder handles.
        """
        # Holds arbitrary JSON-ish values: dicts seed it, nested dict/list
        # members get pushed back for traversal.
        stack: list[object] = [data]
        while stack:
            obj = stack.pop()
            if isinstance(obj, dict):
                items = obj.values()
            elif isinstance(obj, (list, tuple)):
                items = obj
            else:
                continue
            for item in items:
                if JSONSerializer._value_needs_encoding(item):
                    return True
                if isinstance(item, (dict, list)):
                    stack.append(item)
        return False

    @traces(
        spec="004",
        story="US-006",
        reqs=[
            "REQ-004-PERF-001",
            "REQ-004-US-003-005",
            "REQ-004-US-006-001",
            "REQ-004-US-006-002",
            "REQ-004-US-006-025",
            "REQ-004-US-006-026",
        ],
    )
    def to_json(self, packet: CXPPacket, pretty: bool = False) -> str:
        """Serialize a CXPPacket to a JSON string.

        Args:
            packet: The CXP packet to serialize.
            pretty: If True, output indented JSON for readability.

        Returns:
            JSON string representation of the packet.

        Raises:
            SerializationError: If serialization fails.
        """
        try:
            data = self.to_dict(packet)
            indent = 2 if pretty else None
            if self._needs_custom_encoding(data):
                return json.dumps(
                    data,
                    cls=self._encoder,
                    indent=indent,
                    ensure_ascii=False,
                    sort_keys=False,
                )
            else:
                return json.dumps(
                    data,
                    indent=indent,
                    ensure_ascii=False,
                    sort_keys=False,
                )
        except (TypeError, ValueError, OverflowError, RecursionError) as exc:
            raise SerializationError(
                message=f"JSON serialization failed: {exc}",
                format="json",
                details={"packet_id": getattr(packet, "message_id", "unknown")},
            ) from exc

    @traces(
        spec="004",
        story="US-003",
        reqs=[
            "REQ-004-PERF-002",
            "REQ-004-US-003-010",
            "REQ-004-US-006-009",
            "REQ-004-US-006-010",
            "REQ-004-US-006-015",
        ],
    )
    def from_json(self, data: str) -> CXPPacket:
        """Deserialize a JSON string to a CXPPacket.

        Args:
            data: JSON string to deserialize.

        Returns:
            Reconstructed CXPPacket instance.

        Raises:
            DeserializationError: If the JSON is invalid or packet
                reconstruction fails.
        """
        if not isinstance(data, str):
            raise DeserializationError(
                message=f"Expected str, got {type(data).__name__}",
                format="json",
            )

        try:
            parsed = json.loads(data, object_hook=_decode_bytes_hook)
        except (json.JSONDecodeError, ValueError) as exc:
            raise DeserializationError(
                message=f"Invalid JSON: {exc}",
                format="json",
            ) from exc

        if not isinstance(parsed, dict):
            raise DeserializationError(
                message=f"JSON root must be an object, got {type(parsed).__name__}",
                format="json",
            )

        return self.from_dict(parsed)

    @traces(
        spec="004",
        story="US-006",
        reqs=["REQ-004-US-006-002"],
    )
    def to_dict(self, packet: CXPPacket) -> dict[str, Any]:
        """Convert a CXPPacket to a plain dictionary.

        Delegates to packet.to_dict().

        Args:
            packet: The CXP packet to convert.

        Returns:
            Dictionary representation of the packet.

        Raises:
            SerializationError: If conversion fails.
        """
        try:
            return packet.to_dict()
        except Exception as exc:
            raise SerializationError(
                message=f"to_dict failed: {exc}",
                format="json",
                details={"packet_id": getattr(packet, "message_id", "unknown")},
            ) from exc

    @traces(
        spec="004",
        story="US-006",
        reqs=[
            "REQ-004-US-006-010",
            "REQ-004-US-006-016",
        ],
    )
    def from_dict(self, data: dict[str, Any]) -> CXPPacket:
        """Reconstruct a CXPPacket from a dictionary.

        Delegates to CXPPacket.from_dict().

        Args:
            data: Dictionary with packet fields.

        Returns:
            Reconstructed CXPPacket instance.

        Raises:
            DeserializationError: If reconstruction fails.
        """
        if not isinstance(data, dict):
            raise DeserializationError(
                message=f"Expected dict, got {type(data).__name__}",
                format="json",
            )

        try:
            return CXPPacket.from_dict(data)
        except (KeyError, TypeError, ValueError) as exc:
            raise DeserializationError(
                message=f"Packet reconstruction failed: {exc}",
                format="json",
                details={"keys_present": list(data.keys())},
            ) from exc