interactions.py

import json
from typing import List, NamedTuple

from . import BaseObject


class IDNamePair(NamedTuple):
    """Simple type used to help with unpacking event data."""

    id: str
    name: str


class InteractiveEvent(BaseObject):
    response_url: str
    user: IDNamePair
    team: IDNamePair
    channel: IDNamePair

    raw_event: dict

    def __init__(self, event: dict):
        self.raw_event = event
        self.response_url = event["response_url"]


class MessageInteractiveEvent(InteractiveEvent):
    event_type: str
    message_ts: str
    trigger_id: str
    action_id: str
    block_id: str
    message: dict

    def __init__(self, event: dict):
        """Convenience class to parse an interactive message payload from the events API.

        Args:
            event: the raw event dictionary
        """
        super().__init__(event)
        self.user = IDNamePair(event["user"]["id"], event["user"]["username"])
        self.team: IDNamePair = IDNamePair(event["team"]["id"], event["team"]["domain"])
        self.channel: IDNamePair = IDNamePair(event["channel"]["id"], event["channel"]["name"])
        self.event_type = event["type"]
        self.message_ts = event["message"]["ts"]
        self.trigger_id = event["trigger_id"]
        # actions payload is an array, but will only have one item (the action
        # actually interacted with)
        action = event["actions"][0]
        self.action_id = action["action_id"]
        self.block_id = action["block_id"]
        if action.get("selected_option"):
            self.value = action["selected_option"]["value"]
        else:
            self.value = action["value"]
        self.message = event["message"]


class DialogInteractiveEvent(InteractiveEvent):
    event_type: str
    submission: dict
    state: dict

    def __init__(self, event: dict):
        """Convenience class to parse a dialog interaction payload from the events API.

        Args:
            event: the raw event dictionary
        """
        super().__init__(event)
        self.user = IDNamePair(event["user"]["id"], event["user"]["name"])
        self.team = IDNamePair(event["team"]["id"], event["team"]["domain"])
        self.channel = IDNamePair(event["channel"]["id"], event["channel"]["name"])
        self.callback_id = event["callback_id"]
        self.event_type = event["type"]
        self.submission = event["submission"]
        if event["state"]:
            self.state = json.loads(event["state"])
        else:
            self.state = {}

    def require_any(self, requirements: List[str]) -> dict:
        """Convenience method to construct the 'errors' response to send directly back to the invoking HTTP request.

        Args:
          requirements: List of required dialog components, by name
        """
        if any(self.submission.get(requirement, "") for requirement in requirements):  # skipcq: PYL-R1705
            return {}
        else:
            errors = []
            for key in self.submission:
                error_text = "At least one value is required"
                errors.append({"name": key, "error": error_text})
            return {"errors": errors}


class SlashCommandInteractiveEvent(InteractiveEvent):
    trigger_id: str
    command: str
    text: str

    def __init__(self, event: dict):
        """Convenience class to parse a slash command payload from the events API.

        Args:
            event: the raw event dictionary
        """
        super().__init__(event)
        self.user = IDNamePair(event["user_id"], event["user_name"])
        self.channel = IDNamePair(event["channel_id"], event["channel_name"])
        self.team = IDNamePair(event["team_id"], event["team_domain"])
        self.trigger_id = event["trigger_id"]
        self.command = event["command"]
        self.text = event["text"]

    @staticmethod
    def create_reply(message, ephemeral=False) -> dict:
        """Create a reply suitable to send directly back to the invoking HTTP request.

        Args:
          message: Text to send
          ephemeral: Whether the response should be limited to a single user, or to
                broadcast the reply (_and_ the user's original invocation) to the
                channel publicly
        """
        if ephemeral:  # skipcq: PYL-R1705
            return {"text": message, "response_type": "ephemeral"}
        else:
            return {"text": message, "response_type": "in_channel"}