How to Mock a Choreography Saga in Python

import json
from dataclasses import dataclass, asdict
from typing import List, Optional
from enum import Enum


class SagaStatus(Enum):
    PENDING = "PENDING"
    COMPLETING = "COMPLETING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"


@dataclass
class EventEnvelope:
    event_type: str
    order_id: str
    status: SagaStatus
    metadata: Optional[dict] = None

    def to_json(self) -> str:
        return json.dumps(asdict(self))


class ChoreographySagaMock:
    def __init__(self):
        self.events: List[EventEnvelope] = []
        self._status: SagaStatus = SagaStatus.PENDING

    def publish(self, event: EventEnvelope) -> None:
        """Record and print a published event, updating saga status."""
        self.events.append(event)
        if event.status in (SagaStatus.COMPLETING, SagaStatus.COMPLETED):
            self._status = event.status
        elif event.status == SagaStatus.FAILED:
            self._status = SagaStatus.FAILED
        print(f"Event: {event.to_json()}")

    def run(self) -> None:
        """Simulate a full saga choreography with compensating actions."""
        order_id = "ORD-12345"
        self.publish(EventEnvelope("ORDER_CREATED", order_id, SagaStatus.PENDING))
        self.publish(EventEnvelope("INVENTORY_RESERVED", order_id, SagaStatus.COMPLETING))
        self.publish(EventEnvelope("PAYMENT_CHARGED", order_id, SagaStatus.COMPLETING))

        # Simulate a failure mid-saga
        if self._status != SagaStatus.COMPLETED:
            self.publish(EventEnvelope("PAYMENT_FAILED", order_id, SagaStatus.FAILED))
            self.publish(EventEnvelope("INVENTORY_RELEASED", order_id, SagaStatus.PENDING))
            self.publish(EventEnvelope("ORDER_CANCELLED", order_id, SagaStatus.PENDING, {"reason": "payment error"}))

        print(f"Final status: {self._status.value}")
        print(f"Total events: {len(self.events)}")


if __name__ == "__main__":
    saga = ChoreographySagaMock()
    saga.run()