pincer.objects.app.interactions.Interaction.__post_followup_send_handler() - Code Metrics - Inspection of "✨ Guild registration, guild commands fix, command..." - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#158)

unknown

created 2021-10-21 08:53 UTC

Interaction.__post_followup_send_handler() A

↳ Parent: pincer.objects.app.interactions

Complexity

Conditions

Size

Total Lines	18
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	7
dl	0
loc	18
rs	10
c	0
b	0
f	0
cc	2
nop	3

# Copyright Pincer 2021-Present

# Full MIT License can be found in `LICENSE` at the project root.

from __future__ import annotations

from asyncio import gather, iscoroutine, sleep, ensure_future
from dataclasses import dataclass
from enum import IntEnum
from typing import Dict, TYPE_CHECKING

from .command import AppCommandInteractionDataOption, AppCommandOptionType
from .interaction_base import InteractionType, CallbackType
from ..app.select_menu import SelectOption
from ..guild.member import GuildMember
from ..message.context import MessageContext
from ..message.user_message import UserMessage
from ..user import User
from ...utils import APIObject, MISSING, Snowflake, convert

if TYPE_CHECKING:
    from ..message.message import Message
    from ..guild.channel import Channel
    from ..guild.role import Role
    from ...utils import APINullable


class InteractionFlags(IntEnum):
    """
    :param EPHEMERAL:
        only the user receiving the message can see it
    """
    EPHEMERAL = 1 << 6


@dataclass
class ResolvedData(APIObject):
    """
    Represents a Discord Resolved Data structure

    :param users:
        Map of Snowflakes to user objects

    :param members:
        Map of Snowflakes to partial member objects

    :param roles:
        Map of Snowflakes to role objects

    :param channels:
        Map of Snowflakes to partial channel objects

    :param messages:
        Map of Snowflakes to partial message objects
    """
    users: APINullable[Dict[Snowflake, User]] = MISSING
    members: APINullable[Dict[Snowflake, GuildMember]] = MISSING
    roles: APINullable[Dict[Snowflake, Role]] = MISSING
    channels: APINullable[Dict[Snowflake, Channel]] = MISSING
    messages: APINullable[Dict[Snowflake, UserMessage]] = MISSING


@dataclass

class InteractionData(APIObject):
    """
    Represents a Discord Interaction Data structure

    :param id:
        the `ID` of the invoked command

    :param name:
        the `name` of the invoked command

    :param type:
        the `type` of the invoked command

    :param resolved:
        converted users + roles + channels

    :param options:
        the params + values from the user

    :param custom_id:
        the `custom_id` of the component

    :param component_type:
        the type of the component

    :param values:
        the values the user selected

    :param target_id:
        id of the user or message targeted by a user or message command
    """
    id: Snowflake
    name: str
    type: int

    resolved: APINullable[ResolvedData] = MISSING
    options: APINullable[AppCommandInteractionDataOption] = MISSING
    custom_id: APINullable[str] = MISSING
    component_type: APINullable[int] = MISSING
    values: APINullable[SelectOption] = MISSING
    target_id: APINullable[Snowflake] = MISSING

    def __post_init__(self):
        self.id = convert(self.id, Snowflake.from_string)
        self.resolved = convert(
            self.resolved,
            ResolvedData.from_dict,
            ResolvedData
        )
        self.options = convert(
            self.options,
            AppCommandInteractionDataOption.from_dict,
            AppCommandInteractionDataOption
        )
        self.values = convert(
            self.values,
            SelectOption.from_dict,
            SelectOption
        )
        self.target_id = convert(self.target_id, Snowflake.from_string)


@dataclass

class Interaction(APIObject):
    """
    Represents a Discord Interaction object

    :param id:
        id of the interaction

    :param application_id:
        id of the application this interaction is for

    :param type:
        the type of interaction

    :param data:
        the command data payload

    :param guild_id:
        the guild it was sent from

    :param channel_id:
        the channel it was sent from

    :param member:
        guild member data for the invoking user, including permissions

    :param user:
        user object for the invoking user, if invoked in a DM

    :param token:
        a continuation token for responding to the interaction

    :param version:
        read-only property, always `1`

    :param message:
        for components, the message they were attached to
    """
    id: Snowflake
    application_id: Snowflake
    type: InteractionType
    token: str

    version: int = 1
    data: APINullable[InteractionData] = MISSING
    guild_id: APINullable[Snowflake] = MISSING
    channel_id: APINullable[Snowflake] = MISSING
    member: APINullable[GuildMember] = MISSING
    user: APINullable[User] = MISSING
    message: APINullable[UserMessage] = MISSING

    def __post_init__(self):
        self.id = convert(self.id, Snowflake.from_string)
        self.application_id = convert(
            self.application_id, Snowflake.from_string
        )
        self.type = convert(self.type, InteractionType)
        self.data = convert(
            self.data,
            InteractionData.from_dict,
            InteractionData
        )
        self.guild_id = convert(self.guild_id, Snowflake.from_string)
        self.channel_id = convert(self.channel_id, Snowflake.from_string)

        self.member = convert(
            self.member,
            GuildMember.from_dict,
            GuildMember,
            client=self._client
        )

        self.user = convert(
            self.user,
            User.from_dict,
            User,
            client=self._client
        )

        self.message = convert(
            self.message,
            UserMessage.from_dict,
            UserMessage,
            client=self._client
        )

        self._convert_functions = {
            AppCommandOptionType.SUB_COMMAND: None,
            AppCommandOptionType.SUB_COMMAND_GROUP: None,

            AppCommandOptionType.STRING: str,
            AppCommandOptionType.INTEGER: int,
            AppCommandOptionType.BOOLEAN: bool,
            AppCommandOptionType.NUMBER: float,

            AppCommandOptionType.USER: lambda value:
            self._client.get_user(

                convert(value, Snowflake.from_string)
            ),
            AppCommandOptionType.CHANNEL: lambda value:
            self._client.get_channel(

                convert(value, Snowflake.from_string)
            ),
            AppCommandOptionType.ROLE: lambda value:
            self._client.get_role(

                convert(self.guild_id, Snowflake.from_string),
                convert(value, Snowflake.from_string)
            ),
            AppCommandOptionType.MENTIONABLE: None
        }

    async def build(self):
        """
        Sets the parameters in the interaction that need information from the
        discord API.
        """

        if not self.data.options:
            return

        await gather(
            *map(self.convert, self.data.options)
        )

    async def convert(self, option: AppCommandInteractionDataOption):
        """
        Sets an AppCommandInteractionDataOption value paramater to the payload
        type
        """

        converter = self._convert_functions.get(option.type)

        if not converter:
            raise NotImplementedError(
                f"Handling for AppCommandOptionType {option.type} is not "
                "implemented"
            )

        res = converter(option.value)

        if iscoroutine(res):
            option.value = await res
            return

        option.value = res

    def convert_to_message_context(self, command):

        return MessageContext(
            self.id,
            self.member or self.user,
            command,
            self.guild_id,
            self.channel_id
        )

    async def __post_send_handler(self, message: Message):
        """
        Process the interaction after it was sent.

        :param message:
            The interaction message.
        """

        if message.delete_after:
            await sleep(message.delete_after)
            await self.delete()

    def __post_sent(self, message: Message):
        """
        Ensure the `__post_send_handler` method its future.

        :param message:
            The interaction message.
        """
        ensure_future(self.__post_send_handler(message))

    async def reply(self, message: Message) -> UserMessage:
        """
        Initial reply, only works if no ACK has been sent yet.

        :param message:
            The response message!
        """
        content_type, data = message.serialize()

        res = await self._http.patch(
            f"interactions/{self.id}/{self.token}/callback",
            {
                "type": CallbackType.MESSAGE,
                "data": data
            },
            content_type=content_type
        )
        self.__post_sent(message)
        return UserMessage.from_dict(res)

    async def edit(self, message: Message) -> UserMessage:
        """
        Edit an interaction. This is also the way to reply to
        interactions whom have been acknowledged.

        :param message:
            The new message!
        """
        content_type, data = message.serialize()

        resp = await self._http.patch(
            f"webhooks/{self._client.bot.id}/{self.token}/messages/@original",
            data,
            content_type=content_type
        )
        self.__post_sent(message)
        return UserMessage.from_dict(resp)

    async def delete(self):
        """
        Delete the interaction.
        """
        await self._http.delete(
            f"webhooks/{self._client.bot.id}/{self.token}/messages/@original"
        )

    async def __post_followup_send_handler(
            self,
            followup: UserMessage,
            message: Message
    ):
        """
        Process a folloup after it was sent.

        :param followup:
            The followup message that is being post processed.

        :param message:
            The interaction message.
        """

        if message.delete_after:
            await sleep(message.delete_after)
            await self.delete_followup(followup.id)

    def __post_followup_sent(
            self,
            followup: UserMessage,
            message: Message
    ):
        """
        Ensure the `__post_followup_send_handler` method its future.

        :param followup:
            The followup message that is being post processed.

        :param message:
            The followup message.
        """
        ensure_future(self.__post_followup_send_handler(followup, message))

    async def followup(self, message: Message) -> UserMessage:
        """
        Create a follow up message for the interaction.
        This allows you to respond with multiple messages.

        :param message:
            The message!
        """
        content_type, data = message.serialize()

        resp = await self._http.post(
            f"webhooks/{self._client.bot.id}/{self.token}",
            data,
            content_type=content_type
        )
        msg = UserMessage.from_dict(resp)
        self.__post_followup_sent(msg, message)
        return msg

    async def edit_followup(self, message_id: int, message: Message) \
            -> UserMessage:
        """
        Edit a followup message.

        :param message_id:
            The id of the original followup message.

        :param message:
            The new message!
        """
        content_type, data = message.serialize()

        resp = await self._http.patch(
            f"webhooks/{self._client.bot.id}/{self.token}/messages/{message_id}",
            data,
            content_type=content_type
        )
        msg = UserMessage.from_dict(resp)
        self.__post_followup_sent(msg, message)
        return msg

    async def get_followup(self, message_id: int) -> UserMessage:
        """
        Get a followup message by id.

        :param message_id:
            The id of the original followup message that must be fetched.
        """

        resp = await self._http.get(
            f"webhooks/{self._client.bot.id}/{self.token}/messages/{message_id}",
        )
        return UserMessage.from_dict(resp)

    async def delete_followup(self, message_id: int):
        """
        Remove a followup message by id.

        :param message_id:
            The id of the followup message that must be deleted.
        """

        await self._http.delete(
            f"webhooks/{self._client.bot.id}/{self.token}/messages/{message_id}",
        )


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

Pincer-org / Pincer

Pull Request — main (#158)

Interaction.__post_followup_send_handler() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like