Repository URL to install this package:
|
Version:
3.7.0 ▾
|
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( # skipcq: PYL-R1705
self.submission.get(requirement, "") for requirement in requirements
):
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"}