orchestrata_ultimate.core.packet.compat

Reference › orchestrata_ultimate › core › packet › compat

The deprecated backward-compatibility classes of the packet vocabulary. PacketHeader and PacketPayload are the legacy structured representations; their docstrings direct new code to use the flat fields of CXPPacket instead. HeaderProxy and HeadersDict are adapters that let old-style consumers keep reading and writing headers against the flat CXPPacket; the module imports warnings and logging for deprecation reporting. It moved here from cxp with the rest of the packet vocabulary (decision K6) and is re-exported from the packet facade. It imports enums and, for type checking only, the packet itself.

At a glance

Class diagram

Dependency diagram

API

Classes

PacketHeader (dataclass)

Decorators: dataclass · Line 41

CXP packet header.

PacketPayload (dataclass)

Decorators: dataclass · Line 128

CXP packet payload.

HeaderProxy

Line 173

Proxy that maps packet.header.X to packet flat fields.

HeadersDict

Line 331

Dict-like wrapper for packet.headers compatibility.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/packet/compat.py

"""
CXP Packet — deprecated backward-compatibility classes.

Holds the legacy, deprecated packet representations (`PacketHeader`,
`PacketPayload`) and the proxy adapters (`HeaderProxy`, `HeadersDict`) that
let old-style consumers keep working against the flat `CXPPacket` structure.

Lives in ``core.packet`` (Layer 0 vocabulary) alongside the packet itself
(K6: the packet vocabulary moved here from ``cxp`` so that ``core`` is pure).
Re-exported from the ``core.packet`` facade.

Spec: 004-cxp
Author: Mohamed
"""

from __future__ import annotations

import json
import logging
import uuid
import warnings
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet.enums import PacketType, Priority

if TYPE_CHECKING:
    from orchestrata_ultimate.core.packet.packet import CXPPacket

_logger = logging.getLogger(__name__)


# ============================================================================
# Deprecated Data Classes (kept for backward compat with existing consumers)
# ============================================================================


@dataclass
class PacketHeader:
    """CXP packet header.

    DEPRECATED: Use flat CXPPacket fields directly.
    Kept for backward compatibility with existing consumers.
    """

    # Core fields
    packet_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    type: PacketType = PacketType.MESSAGE
    version: str = "1.0"

    # Routing
    sender: str = ""
    recipient: str = ""
    reply_to: str | None = None

    # Delivery
    priority: Priority = Priority.NORMAL
    ttl: int = 3600
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    expires_at: datetime | None = None

    # Metadata
    correlation_id: str | None = None
    conversation_id: str | None = None
    tags: list[str] = field(default_factory=list)

    # Security
    signature: str | None = None
    encryption_key_id: str | None = None

    def __post_init__(self) -> None:
        warnings.warn(
            "PacketHeader is deprecated. Use flat CXPPacket fields directly.",
            DeprecationWarning,
            stacklevel=3,
        )
        if self.expires_at is None and self.ttl > 0:
            self.expires_at = self.created_at + timedelta(seconds=self.ttl)

    @traces(
        spec="004",
        story="US-006",
        reqs=["REQ-004-US-006-002", "REQ-004-US-006-003", "REQ-004-US-006-004"],
    )
    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["type"] = self.type.value
        data["priority"] = self.priority.value
        data["created_at"] = self.created_at.isoformat()
        if self.expires_at:
            data["expires_at"] = self.expires_at.isoformat()
        return {k: v for k, v in data.items() if v is not None}

    @traces(
        spec="004",
        story="US-006",
        reqs=[
            "REQ-004-US-006-011",
            "REQ-004-US-006-012",
        ],
    )
    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> PacketHeader:
        if "type" in data and isinstance(data["type"], str):
            data["type"] = PacketType(data["type"])
        if "priority" in data and isinstance(data["priority"], (int, str)):
            if isinstance(data["priority"], str):
                data["priority"] = Priority(int(data["priority"]))
            else:
                try:
                    data["priority"] = Priority(data["priority"])
                except ValueError:
                    data["priority"] = Priority.NORMAL
        if "created_at" in data and isinstance(data["created_at"], str):
            data["created_at"] = datetime.fromisoformat(
                data["created_at"].replace("Z", "+00:00")
            )
        if "expires_at" in data and isinstance(data["expires_at"], str):
            data["expires_at"] = datetime.fromisoformat(
                data["expires_at"].replace("Z", "+00:00")
            )
        return cls(**data)


@dataclass
class PacketPayload:
    """CXP packet payload.

    DEPRECATED: Use flat CXPPacket.payload (dict|bytes) directly.
    Kept for backward compatibility with existing consumers.
    """

    content_type: str = "application/json"
    body: Any = None
    attachments: list[dict[str, Any]] = field(default_factory=list)
    context: dict[str, Any] = field(default_factory=dict)
    references: list[str] = field(default_factory=list)
    language: str = "en"
    encoding: str = "utf-8"

    def __post_init__(self) -> None:
        warnings.warn(
            "PacketPayload is deprecated. Use flat CXPPacket.payload (dict|bytes) directly.",
            DeprecationWarning,
            stacklevel=3,
        )

    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        if self.body is not None and not isinstance(
            self.body, (str, int, float, bool, type(None))
        ):
            data["body"] = json.dumps(self.body, default=str)
        return data

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> PacketPayload:
        if "body" in data and isinstance(data["body"], str):
            try:
                data["body"] = json.loads(data["body"])
            except json.JSONDecodeError as exc:
                _logger.debug("parse body JSON: %s", exc)
        return cls(**data)


# ============================================================================
# Backward Compat Proxies
# ============================================================================


class HeaderProxy:
    """Proxy that maps packet.header.X to packet flat fields."""

    __slots__ = ("_packet",)

    _packet: CXPPacket

    def __init__(self, packet: CXPPacket) -> None:
        object.__setattr__(self, "_packet", packet)

    @property
    def sender(self) -> str:
        return self._packet.from_agent

    @sender.setter
    def sender(self, value: str) -> None:
        self._packet.from_agent = value

    @property
    def recipient(self) -> str:
        return self._packet.to_agent

    @recipient.setter
    def recipient(self, value: str) -> None:
        self._packet.to_agent = value

    @property
    def packet_id(self) -> str:
        return self._packet.message_id

    @packet_id.setter
    def packet_id(self, value: str) -> None:
        self._packet.message_id = value

    @property
    def type(self) -> PacketType:
        return self._packet.packet_type

    @type.setter
    def type(self, value: PacketType) -> None:
        self._packet.packet_type = value

    @property
    def priority(self) -> Priority:
        try:
            return Priority(self._packet.priority)
        except ValueError:
            return Priority.NORMAL

    @priority.setter
    def priority(self, value: Priority | int) -> None:
        if isinstance(value, Priority):
            self._packet.priority = value.value
        else:
            self._packet.priority = int(value)

    @property
    def created_at(self) -> datetime:
        return self._packet.timestamp

    @created_at.setter
    def created_at(self, value: datetime) -> None:
        self._packet.timestamp = value

    @property
    def expires_at(self) -> datetime | None:
        return self._packet.expires_at

    @expires_at.setter
    def expires_at(self, value: datetime | None) -> None:
        self._packet.expires_at = value

    @property
    def correlation_id(self) -> str | None:
        return self._packet.correlation_id

    @correlation_id.setter
    def correlation_id(self, value: str | None) -> None:
        self._packet.correlation_id = value

    @property
    def conversation_id(self) -> str | None:
        return self._packet.thread_id

    @conversation_id.setter
    def conversation_id(self, value: str | None) -> None:
        self._packet.thread_id = value or ""

    @property
    def ttl(self) -> int:
        return self._packet.ttl_seconds

    @ttl.setter
    def ttl(self, value: int) -> None:
        self._packet.ttl_seconds = value

    @property
    def version(self) -> str:
        return self._packet.protocol_version

    @property
    def reply_to(self) -> str | None:
        return self._packet.reply_to

    @reply_to.setter
    def reply_to(self, value: str | None) -> None:
        self._packet.reply_to = value

    @property
    def tags(self) -> list[str]:
        return self._packet.tags

    @property
    def signature(self) -> str | None:
        return self._packet.metadata.get("signature")

    @property
    def encryption_key_id(self) -> str | None:
        return self._packet.metadata.get("encryption_key_id")

    @traces(
        spec="004",
        story="US-006",
        reqs=["REQ-004-US-006-003", "REQ-004-US-006-004"],
    )
    def to_dict(self) -> dict[str, Any]:
        """Serialize header fields to dict."""
        data: dict[str, Any] = {
            "packet_id": self.packet_id,
            "type": self.type.value,
            "version": self.version,
            "sender": self.sender,
            "recipient": self.recipient,
            "priority": self.priority.value,
            "ttl": self.ttl,
            "created_at": (
                self.created_at.isoformat()
                if isinstance(self.created_at, datetime)
                else str(self.created_at)
            ),
        }
        if self.expires_at:
            data["expires_at"] = (
                self.expires_at.isoformat()
                if isinstance(self.expires_at, datetime)
                else str(self.expires_at)
            )
        if self.correlation_id:
            data["correlation_id"] = self.correlation_id
        if self.conversation_id:
            data["conversation_id"] = self.conversation_id
        if self.reply_to:
            data["reply_to"] = self.reply_to
        if self.tags:
            data["tags"] = self.tags
        return data


class HeadersDict:
    """Dict-like wrapper for packet.headers compatibility."""

    __slots__ = ("_packet",)

    _packet: CXPPacket

    def __init__(self, packet: CXPPacket) -> None:
        object.__setattr__(self, "_packet", packet)

    @traces(
        spec="004",
        story="US-002",
        reqs=[
            "REQ-004-US-002-001",
            "REQ-004-US-002-002",
            "REQ-004-US-002-003",
            "REQ-004-US-002-004",
            "REQ-004-US-002-005",
            "REQ-004-US-002-006",
            "REQ-004-US-002-007",
            "REQ-004-US-002-008",
        ],
    )
    def get(self, key: str, default: Any = None) -> Any:
        mapping = {
            "PACKET_ID": self._packet.message_id,
            "TO": self._packet.to_agent,
            "FROM": self._packet.from_agent,
            "MESSAGE_ID": self._packet.message_id,
            "SUBJECT": self._packet.subject,
            "THREAD_ID": self._packet.thread_id,
            "PRIORITY": self._packet.priority,
            "TYPE": self._packet.packet_type,
            "TIMESTAMP": self._packet.timestamp,
        }
        return mapping.get(key, default)

    def __getitem__(self, key: str) -> Any:
        val = self.get(key)
        if val is None:
            raise KeyError(key)
        return val

    def __contains__(self, key: str) -> bool:
        return key in (
            "PACKET_ID",
            "TO",
            "FROM",
            "MESSAGE_ID",
            "SUBJECT",
            "THREAD_ID",
            "PRIORITY",
            "TYPE",
            "TIMESTAMP",
        )