orchestrata_ultimate.cxp.serializer

Reference › orchestrata_ultimate › cxp › serializer

CXPSerializer is the original, simple serializer. It converts a CXPPacket to an indented JSON string with to_json, back with from_json, and to and from dictionaries with to_dict and from_dict. The older names serialize, deserialize, serialize_to_dict and deserialize_from_dict are kept as backward-compatible aliases. It does no custom encoding and raises no CXP-specific errors; the fuller implementation is json_serializer, with binary_serializer for bytes. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPSerializer

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

Serializes and deserializes CXP packets.

Source

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

"""
CXP Serializer - Serializes and deserializes CXP packets.

Handles JSON serialization for network transmission and storage.

Spec: 004-cxp
Author: Mohamed
"""

import json
import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket

logger = logging.getLogger(__name__)


@traces(
    spec="004",
    story="US-006",
    reqs=[
        "REQ-004-US-006-001",
        "REQ-004-US-006-009",
        "REQ-004-US-006-025",
    ],
)
class CXPSerializer:
    """Serializes and deserializes CXP packets.

    Methods:
        to_json / from_json: Spec 004 names
        serialize / deserialize: Backward compat aliases
    """

    def __init__(self, config: dict[str, Any] | None = None) -> None:
        self._config = config or {}
        logger.info("CXPSerializer initialized")

    @traces(
        spec="004",
        story="US-006",
        reqs=["REQ-004-US-006-001", "REQ-004-US-006-025"],
    )
    def to_json(self, packet: CXPPacket) -> str:
        """Serialize packet to JSON string."""
        data = packet.to_dict()
        return json.dumps(data, indent=2, default=str)

    @traces(spec="004", story="US-006", reqs=["REQ-004-US-006-009"])
    def from_json(self, data: str) -> CXPPacket:
        """Deserialize JSON string to packet."""
        parsed = json.loads(data)
        return CXPPacket.from_dict(parsed)

    def to_dict(self, packet: CXPPacket) -> dict[str, Any]:
        """Serialize packet to dictionary."""
        return packet.to_dict()

    def from_dict(self, data: dict[str, Any]) -> CXPPacket:
        """Deserialize dictionary to packet."""
        return CXPPacket.from_dict(data)

    # Backward compat aliases
    def serialize(self, packet: CXPPacket) -> str:
        """Deprecated: use to_json."""
        return self.to_json(packet)

    def deserialize(self, data: str) -> CXPPacket:
        """Deprecated: use from_json."""
        return self.from_json(data)

    def serialize_to_dict(self, packet: CXPPacket) -> dict[str, Any]:
        """Deprecated: use to_dict."""
        return self.to_dict(packet)

    def deserialize_from_dict(self, data: dict[str, Any]) -> CXPPacket:
        """Deprecated: use from_dict."""
        return self.from_dict(data)