Passed
Pull Request — main (#158)
by
unknown
03:42
created

pincer.objects.app.interactions   A

Complexity

Total Complexity 23

Size/Duplication

Total Lines 420
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
wmc 23
eloc 181
dl 0
loc 420
rs 10
c 0
b 0
f 0

15 Methods

Rating   Name   Duplication   Size   Complexity  
A Interaction.edit() 0 17 1
A Interaction.delete() 0 6 1
B Interaction.__post_init__() 0 58 4
A Interaction.get_followup() 0 12 1
A Interaction.__post_sent() 0 8 1
A Interaction.delete_followup() 0 10 1
A Interaction.convert() 0 21 3
A Interaction.followup() 0 18 1
A Interaction.convert_to_message_context() 0 7 1
A Interaction.__post_send_handler() 0 11 2
A Interaction.build() 0 11 2
A Interaction.__post_followup_send_handler() 0 18 2
A Interaction.reply() 0 18 1
A Interaction.edit_followup() 0 21 1
A Interaction.__post_followup_sent() 0 15 1
1
# Copyright Pincer 2021-Present
0 ignored issues
show
introduced by
Missing module docstring
Loading history...
2
# Full MIT License can be found in `LICENSE` at the project root.
3
4
from __future__ import annotations
5
6
from asyncio import gather, iscoroutine, sleep, ensure_future
7
from dataclasses import dataclass
8
from typing import Dict, TYPE_CHECKING
9
10
from .command_types import AppCommandOptionType
11
from .interaction_base import InteractionType, CallbackType
12
from ..app.select_menu import SelectOption
13
from ..guild.member import GuildMember
14
from ..message.context import MessageContext
15
from ..message.user_message import UserMessage
16
from ..user import User
17
from ...utils import APIObject, convert
18
from ...utils.snowflake import Snowflake
19
from ...utils.types import MISSING
20
21
if TYPE_CHECKING:
22
    from .command import AppCommandInteractionDataOption
23
    from ..message.message import Message
24
    from ..guild.channel import Channel
0 ignored issues
show
introduced by
Cannot import 'guild.channel' due to syntax error 'invalid syntax (<unknown>, line 235)'
Loading history...
Bug introduced by
The name channel does not seem to exist in module pincer.objects.guild.
Loading history...
25
    from ..guild.role import Role
26
    from ...utils import APINullable
27
28
29
@dataclass
30
class ResolvedData(APIObject):
31
    """
32
    Represents a Discord Resolved Data structure
33
34
    :param users:
35
        Map of Snowflakes to user objects
36
37
    :param members:
38
        Map of Snowflakes to partial member objects
39
40
    :param roles:
41
        Map of Snowflakes to role objects
42
43
    :param channels:
44
        Map of Snowflakes to partial channel objects
45
46
    :param messages:
47
        Map of Snowflakes to partial message objects
48
    """
49
    users: APINullable[Dict[Snowflake, User]] = MISSING
50
    members: APINullable[Dict[Snowflake, GuildMember]] = MISSING
51
    roles: APINullable[Dict[Snowflake, Role]] = MISSING
52
    channels: APINullable[Dict[Snowflake, Channel]] = MISSING
53
    messages: APINullable[Dict[Snowflake, UserMessage]] = MISSING
54
55
56
@dataclass
0 ignored issues
show
best-practice introduced by
Too many instance attributes (9/7)
Loading history...
57
class InteractionData(APIObject):
58
    """
59
    Represents a Discord Interaction Data structure
60
61
    :param id:
62
        the `ID` of the invoked command
63
64
    :param name:
65
        the `name` of the invoked command
66
67
    :param type:
68
        the `type` of the invoked command
69
70
    :param resolved:
71
        converted users + roles + channels
72
73
    :param options:
74
        the params + values from the user
75
76
    :param custom_id:
77
        the `custom_id` of the component
78
79
    :param component_type:
80
        the type of the component
81
82
    :param values:
83
        the values the user selected
84
85
    :param target_id:
86
        id of the user or message targeted by a user or message command
87
    """
88
    id: Snowflake
89
    name: str
90
    type: int
91
92
    resolved: APINullable[ResolvedData] = MISSING
93
    options: APINullable[AppCommandInteractionDataOption] = MISSING
94
    custom_id: APINullable[str] = MISSING
95
    component_type: APINullable[int] = MISSING
96
    values: APINullable[SelectOption] = MISSING
97
    target_id: APINullable[Snowflake] = MISSING
98
99
100
@dataclass
0 ignored issues
show
best-practice introduced by
Too many instance attributes (12/7)
Loading history...
101
class Interaction(APIObject):
102
    """
103
    Represents a Discord Interaction object
104
105
    :param id:
106
        id of the interaction
107
108
    :param application_id:
109
        id of the application this interaction is for
110
111
    :param type:
112
        the type of interaction
113
114
    :param data:
115
        the command data payload
116
117
    :param guild_id:
118
        the guild it was sent from
119
120
    :param channel_id:
121
        the channel it was sent from
122
123
    :param member:
124
        guild member data for the invoking user, including permissions
125
126
    :param user:
127
        user object for the invoking user, if invoked in a DM
128
129
    :param token:
130
        a continuation token for responding to the interaction
131
132
    :param version:
133
        read-only property, always `1`
134
135
    :param message:
136
        for components, the message they were attached to
137
    """
138
    id: Snowflake
139
    application_id: Snowflake
140
    type: InteractionType
141
    token: str
142
143
    version: int = 1
144
    data: APINullable[InteractionData] = MISSING
145
    guild_id: APINullable[Snowflake] = MISSING
146
    channel_id: APINullable[Snowflake] = MISSING
147
    member: APINullable[GuildMember] = MISSING
148
    user: APINullable[User] = MISSING
149
    message: APINullable[UserMessage] = MISSING
150
151
    def __post_init__(self):
152
        self.id = convert(self.id, Snowflake.from_string)
153
        self.application_id = convert(
154
            self.application_id, Snowflake.from_string
155
        )
156
        self.type = convert(self.type, InteractionType)
157
        self.data = convert(
158
            self.data,
159
            InteractionData.from_dict,
160
            InteractionData
161
        )
162
        self.guild_id = convert(self.guild_id, Snowflake.from_string)
163
        self.channel_id = convert(self.channel_id, Snowflake.from_string)
164
165
        self.member = convert(
166
            self.member,
167
            GuildMember.from_dict,
168
            GuildMember,
169
            client=self._client
170
        )
171
172
        self.user = convert(
173
            self.user,
174
            User.from_dict,
175
            User,
176
            client=self._client
177
        )
178
179
        self.message = convert(
180
            self.message,
181
            UserMessage.from_dict,
182
            UserMessage,
183
            client=self._client
184
        )
185
186
        self._convert_functions = {
187
            AppCommandOptionType.SUB_COMMAND: None,
188
            AppCommandOptionType.SUB_COMMAND_GROUP: None,
189
190
            AppCommandOptionType.STRING: str,
191
            AppCommandOptionType.INTEGER: int,
192
            AppCommandOptionType.BOOLEAN: bool,
193
            AppCommandOptionType.NUMBER: float,
194
195
            AppCommandOptionType.USER: lambda value:
196
            self._client.get_user(
0 ignored issues
show
Coding Style introduced by
Wrong continued indentation (add 27 spaces).
Loading history...
197
                convert(value, Snowflake.from_string)
198
            ),
199
            AppCommandOptionType.CHANNEL: lambda value:
200
            self._client.get_channel(
0 ignored issues
show
Coding Style introduced by
Wrong continued indentation (add 30 spaces).
Loading history...
201
                convert(value, Snowflake.from_string)
202
            ),
203
            AppCommandOptionType.ROLE: lambda value:
204
            self._client.get_role(
0 ignored issues
show
Coding Style introduced by
Wrong continued indentation (add 27 spaces).
Loading history...
205
                convert(self.guild_id, Snowflake.from_string),
206
                convert(value, Snowflake.from_string)
207
            ),
208
            AppCommandOptionType.MENTIONABLE: None
209
        }
210
211
    async def build(self):
212
        """
213
        Sets the parameters in the interaction that need information from the
214
        discord API.
215
        """
216
217
        if not self.data.options:
218
            return
219
220
        await gather(
221
            *map(self.convert, self.data.options)
222
        )
223
224
    async def convert(self, option: AppCommandInteractionDataOption):
225
        """
226
        Sets an AppCommandInteractionDataOption value paramater to the payload
227
        type
228
        """
229
230
        converter = self._convert_functions.get(option.type)
231
232
        if not converter:
233
            raise NotImplementedError(
234
                f"Handling for AppCommandOptionType {option.type} is not "
235
                "implemented"
236
            )
237
238
        res = converter(option.value)
239
240
        if iscoroutine(res):
241
            option.value = await res
242
            return
243
244
        option.value = res
245
246
    def convert_to_message_context(self, command):
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
247
        return MessageContext(
248
            self.id,
249
            self.member or self.user,
250
            command,
251
            self.guild_id,
252
            self.channel_id
253
        )
254
255
    async def __post_send_handler(self, message: Message):
256
        """
257
        Process the interaction after it was sent.
258
259
        :param message:
260
            The interaction message.
261
        """
262
263
        if message.delete_after:
264
            await sleep(message.delete_after)
265
            await self.delete()
266
267
    def __post_sent(self, message: Message):
268
        """
269
        Ensure the `__post_send_handler` method its future.
270
271
        :param message:
272
            The interaction message.
273
        """
274
        ensure_future(self.__post_send_handler(message))
275
276
    async def reply(self, message: Message):
277
        """
278
        Initial reply, only works if no ACK has been sent yet.
279
280
        :param message:
281
            The response message!
282
        """
283
        content_type, data = message.serialize()
284
285
        await self._http.post(
286
            f"interactions/{self.id}/{self.token}/callback",
287
            {
288
                "type": CallbackType.MESSAGE,
289
                "data": data
290
            },
291
            content_type=content_type
292
        )
293
        self.__post_sent(message)
294
295
    async def edit(self, message: Message) -> UserMessage:
296
        """
297
        Edit an interaction. This is also the way to reply to
298
        interactions whom have been acknowledged.
299
300
        :param message:
301
            The new message!
302
        """
303
        content_type, data = message.serialize()
304
305
        resp = await self._http.patch(
306
            f"webhooks/{self._client.bot.id}/{self.token}/messages/@original",
307
            data,
308
            content_type=content_type
309
        )
310
        self.__post_sent(message)
311
        return UserMessage.from_dict(resp)
312
313
    async def delete(self):
314
        """
315
        Delete the interaction.
316
        """
317
        await self._http.delete(
318
            f"webhooks/{self._client.bot.id}/{self.token}/messages/@original"
319
        )
320
321
    async def __post_followup_send_handler(
322
            self,
323
            followup: UserMessage,
324
            message: Message
325
    ):
326
        """
327
        Process a folloup after it was sent.
328
329
        :param followup:
330
            The followup message that is being post processed.
331
332
        :param message:
333
            The interaction message.
334
        """
335
336
        if message.delete_after:
337
            await sleep(message.delete_after)
338
            await self.delete_followup(followup.id)
339
340
    def __post_followup_sent(
341
            self,
342
            followup: UserMessage,
343
            message: Message
344
    ):
345
        """
346
        Ensure the `__post_followup_send_handler` method its future.
347
348
        :param followup:
349
            The followup message that is being post processed.
350
351
        :param message:
352
            The followup message.
353
        """
354
        ensure_future(self.__post_followup_send_handler(followup, message))
355
356
    async def followup(self, message: Message) -> UserMessage:
357
        """
358
        Create a follow up message for the interaction.
359
        This allows you to respond with multiple messages.
360
361
        :param message:
362
            The message!
363
        """
364
        content_type, data = message.serialize()
365
366
        resp = await self._http.post(
367
            f"webhooks/{self._client.bot.id}/{self.token}",
368
            data,
369
            content_type=content_type
370
        )
371
        msg = UserMessage.from_dict(resp)
372
        self.__post_followup_sent(msg, message)
373
        return msg
374
375
    async def edit_followup(self, message_id: int, message: Message) \
376
            -> UserMessage:
377
        """
378
        Edit a followup message.
379
380
        :param message_id:
381
            The id of the original followup message.
382
383
        :param message:
384
            The new message!
385
        """
386
        content_type, data = message.serialize()
387
388
        resp = await self._http.patch(
389
            f"webhooks/{self._client.bot.id}/{self.token}/messages/{message_id}",
390
            data,
391
            content_type=content_type
392
        )
393
        msg = UserMessage.from_dict(resp)
394
        self.__post_followup_sent(msg, message)
395
        return msg
396
397
    async def get_followup(self, message_id: int) -> UserMessage:
398
        """
399
        Get a followup message by id.
400
401
        :param message_id:
402
            The id of the original followup message that must be fetched.
403
        """
404
405
        resp = await self._http.get(
406
            f"webhooks/{self._client.bot.id}/{self.token}/messages/{message_id}",
407
        )
408
        return UserMessage.from_dict(resp)
409
410
    async def delete_followup(self, message_id: int):
411
        """
412
        Remove a followup message by id.
413
414
        :param message_id:
415
            The id of the followup message that must be deleted.
416
        """
417
418
        await self._http.delete(
419
            f"webhooks/{self._client.bot.id}/{self.token}/messages/{message_id}",
420
        )
421