interaction¶

class Event(BaseModel):
    """
    An event is a message that is sent from one worker inside the Automa to the application layer outside the Automa.
    """
    event_type: Optional[str] = None
    """The type of the event. The type of the event is used to identify the event handler registered to handle the event."""
    timestamp: datetime = datetime.now()
    """The timestamp of the event."""
    data: Optional[Any] = None
    """The data attached to the event."""

class Feedback(BaseModel):
    """
    A feedback is a message that is sent from the application layer outside the Automa to a worker inside the Automa.
    """
    data: Any
    """The data attached to the feedback."""

class FeedbackSender(ABC):
    """
    The appliction layer must use `FeedbackSender` to send back feedback to the worker inside the Automa.
    """
    @abstractmethod
    def send(self, feedback: Feedback) -> None:
        """
        Send feedback to the Automa.
        This method can be called only once for each event.

        This `send` method can be safely called in several different scenarios:
        - In the same asyncio Task of the same event loop as the event handler.
        - In a different asyncio Task of the same event loop as the event handler.
        - In a different thread from the event handler.

        Parameters
        ----------
        feedback: Feedback
            The feedback to be sent.
        """
        ...

class InteractionFeedback(Feedback):
    """
    A feedback object that contains both the data provided by the user and the `interaction_id`, which uniquely identifies the corresponding interaction.
    """
    interaction_id: str
    """ The unique identifier for the interaction."""
    timestamp: datetime = datetime.now()
    """The timestamp of the feedback."""

class InteractionException(Exception):
    """
    An exception raised when the `interact_with_human` method is called one or more times within the latest event loop iteration, causing one or multiple human interactions to be triggered.
    """
    _interactions: List[Interaction]
    """The list of interactions that occurred during the latest event loop iteration."""
    _snapshot: "Snapshot"
    """The snapshot of the Automa's current state."""

    def __init__(self, interactions: List[Interaction], snapshot: "Snapshot"):
        self._interactions = interactions
        self._snapshot = snapshot

    @property
    def interactions(self) -> List[Interaction]:
        """
        A list of `Interaction` objects that occurred during the latest event loop iteration.

        Multiple `Interaction` objects may be generated because, within the latest event loop iteration, multiple workers calling the `interact_with_human` method might be running concurrently in parallel branches of the graph.
        """
        return self._interactions

    @property
    def snapshot(self) -> "Snapshot":
        """
        A `Snapshot` of the Automa's current state.
        The serialization is triggered automatically by the `interact_with_human` method.
        """
        return self._snapshot

class Interaction(BaseModel):
    """
    An object that represents a single interaction between the Automa and a human. 
    Each call to `interact_with_human` will generate an `Interaction` object which will be included in the `InteractionException` raised.
    """
    interaction_id: str
    """ The unique identifier for the interaction."""
    event: Event
    """The event that triggered the interaction."""