forked from slackapi/python-slack-sdk
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathinteractions.py
More file actions
141 lines (117 loc) · 4.43 KB
/
Copy pathinteractions.py
File metadata and controls
141 lines (117 loc) · 4.43 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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"}