orchestrata_ultimate.core.packet.results

Reference › orchestrata_ultimate › core › packet › results

The supporting result dataclasses of the CXP packet (Spec 004), a leaf module with no dependency on CXPPacket. DeliveryResult reports a delivery attempt: success flag, packet id, destination, timestamp, and optional error message, error code, retry-after and latency. QueueStatus summarises a queue: name, total, pending, in-flight and failed message counts, average delivery time, success rate and the last delivery and failure times. ValidationResult reports packet validation: a valid flag, the packet id, error and warning lists, a timestamp and the validator version. Each has to_dict and from_dict; DeliveryResult.to_dict drops fields that are None. packet imports them.

At a glance

Class diagram

Dependency diagram

API

Classes

DeliveryResult (dataclass)

Decorators: dataclass · Line 25

Result of packet delivery attempt.

QueueStatus (dataclass)

Decorators: dataclass · Line 60

Status of message queue.

ValidationResult (dataclass)

Decorators: traces(spec='004', story='US-007', reqs=['REQ-004-US-007-012']), dataclass · Line 105

Result of packet validation.

Source

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

# mypy: ignore-errors
"""CXP result/value dataclasses.

Leaf, cycle-free cluster extracted from ``_packet_impl``: the supporting
result dataclasses (DeliveryResult, QueueStatus, ValidationResult). These
have no dependency on CXPPacket or the legacy proxies.

Spec: 004-cxp
"""

from __future__ import annotations

from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

# ============================================================================
# Supporting Data Classes (unchanged from original)
# ============================================================================


@dataclass
class DeliveryResult:
    """Result of packet delivery attempt."""

    success: bool
    packet_id: str
    destination: str
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))

    error_message: str | None = None
    error_code: str | None = None
    retry_after: int | None = None
    delivery_latency: float | None = None

    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["timestamp"] = self.timestamp.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",
        ],
    )
    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> DeliveryResult:
        if "timestamp" in data and isinstance(data["timestamp"], str):
            data["timestamp"] = datetime.fromisoformat(
                data["timestamp"].replace("Z", "+00:00")
            )
        return cls(**data)


@dataclass
class QueueStatus:
    """Status of message queue."""

    queue_name: str
    total_messages: int
    pending_messages: int
    in_flight_messages: int
    failed_messages: int

    avg_delivery_time: float | None = None
    delivery_success_rate: float | None = None

    last_delivery: datetime | None = None
    last_failure: datetime | None = None

    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        if self.last_delivery:
            data["last_delivery"] = self.last_delivery.isoformat()
        if self.last_failure:
            data["last_failure"] = self.last_failure.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",
        ],
    )
    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> QueueStatus:
        if "last_delivery" in data and isinstance(data["last_delivery"], str):
            data["last_delivery"] = datetime.fromisoformat(
                data["last_delivery"].replace("Z", "+00:00")
            )
        if "last_failure" in data and isinstance(data["last_failure"], str):
            data["last_failure"] = datetime.fromisoformat(
                data["last_failure"].replace("Z", "+00:00")
            )
        return cls(**data)


@traces(spec="004", story="US-007", reqs=["REQ-004-US-007-012"])
@dataclass
class ValidationResult:
    """Result of packet validation."""

    valid: bool
    packet_id: str

    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)

    validated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    validator_version: str = "1.0"

    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["validated_at"] = self.validated_at.isoformat()
        return data

    @traces(
        spec="004",
        story="US-006",
        reqs=[
            "REQ-004-US-006-011",
        ],
    )
    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ValidationResult:
        if "validated_at" in data and isinstance(data["validated_at"], str):
            data["validated_at"] = datetime.fromisoformat(
                data["validated_at"].replace("Z", "+00:00")
            )
        return cls(**data)