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

Passed

Pull Request — main (#158)

unknown

created 2021-10-25 18:00 UTC

Interaction.__post_send_handler() A

↳ Parent: pincer.objects.app.interactions

Complexity

Conditions

Size

Total Lines	12
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	12
rs	10
c	0
b	0
f	0
cc	2
nop	2

# 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 typing import Dict, TYPE_CHECKING

from .command_types import 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, convert
from ...utils.snowflake import Snowflake
from ...utils.types import MISSING

if TYPE_CHECKING:
    from .command import AppCommandInteractionDataOption
    from ..message.message import Message
    from ..guild.channel import Channel

    from ..guild.role import Role
    from ...utils import APINullable


@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


@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):
        """|coro|

        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):
        """|coro|

        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.

        Parameters
        ----------
        message :class:`~.pincer.objects.message.message.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.

        Parameters
        ----------
        message :class:`~.pincer.objects.message.message.Message`
            The interaction message.
        """
        ensure_future(self.__post_send_handler(message))

    async def reply(self, message: Message):
        """|coro|

        Initial reply, only works if no ACK has been sent yet.

        Parameters
        ----------
        message :class:`~.pincer.objects.message.message.Message`
            The response message!
        """
        content_type, data = message.serialize()

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

    async def edit(self, message: Message) -> UserMessage:
        """|coro|

        Edit an interaction. This is also the way to reply to
        interactions whom have been acknowledged.

        Parameters
        ----------
        message :class:`~.pincer.objects.message.message.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):
        """|coro|

        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.

        Parameters
        ----------
        followup :class:`~.pincer.objects.message.user_message.UserMessage`
            The followup message that is being post processed.
        message :class:`~.pincer.objects.message.message.Message`
            The followup 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.

        Parameters
        ----------
        followup :class:`~.pincer.objects.message.user_message.UserMessage`
            The followup message that is being post processed.
        message :class:`~.pincer.objects.message.message.Message`
            The followup message.
        """
        ensure_future(self.__post_followup_send_handler(followup, message))

    async def followup(self, message: Message) -> UserMessage:
        """|coro|

        Create a follow up message for the interaction.
        This allows you to respond with multiple messages.

        Parameters
        ----------
        message :class:`~.pincer.objects.message.message.Message`
            The message to sent.

        Returns
        -------
        :class:`~.pincer.objects.message.user_message.UserMessage`
            The message that has been sent.
        """
        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:
        """|coro|

        Edit a followup message.

        Parameters
        ----------
        message_id :class:`int`
            The id of the original followup message.
        message :class:`~.pincer.objects.message.message.Message`
            The message to edit.

        Returns
        -------
        :class:`~.pincer.objects.message.user_message.UserMessage`
            The updated message object.
        """
        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:
        """|coro|

        Get a followup message by id.

        Parameters
        ----------
        message_id :class:`int`
            The id of the original followup message that must be fetched.

        Returns
        -------
        :class:`~.pincer.objects.message.user_message.UserMessage`
            The fetched message object.
        """

        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):
        """|coro|

        Remove a followup message by id.

        Parameters
        ----------
        message_id :class:`int`
            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 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 2021-10-22 14:25 UTC by Report Bug Copy Issue Report Cannot import 'guild.channel' due to syntax error 'invalid syntax (<unknown>, line 235)' Loading history... Bug introduced 2021-10-22 14:25 UTC by Report Bug Copy Issue Report 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 2021-09-07 15:34 UTC by Report Bug Copy Issue Report 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 2021-10-14 09:19 UTC by Report Bug Copy Issue Report 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 2021-10-21 08:55 UTC by Report Bug Copy Issue Report 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 2021-10-21 08:55 UTC by Report Bug Copy Issue Report 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 2021-10-21 08:55 UTC by Report Bug Copy Issue Report 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			"""\|coro\|
213
214			Sets the parameters in the interaction that need information
215			from the discord API.
216			"""
217
218			if not self.data.options:
219			return
220
221			await gather(
222			*map(self.convert, self.data.options)
223			)
224
225			async def convert(self, option: AppCommandInteractionDataOption):
226			"""\|coro\|
227
228			Sets an AppCommandInteractionDataOption value paramater to
229			the payload type
230			"""
231
232			converter = self._convert_functions.get(option.type)
233
234			if not converter:
235			raise NotImplementedError(
236			f"Handling for AppCommandOptionType {option.type} is not "
237			"implemented"
238			)
239
240			res = converter(option.value)
241
242			if iscoroutine(res):
243			option.value = await res
244			return
245
246			option.value = res
247
248			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...
249			return MessageContext(
250			self.id,
251			self.member or self.user,
252			command,
253			self.guild_id,
254			self.channel_id
255			)
256
257			async def __post_send_handler(self, message: Message):
258			"""Process the interaction after it was sent.
259
260			Parameters
261			----------
262			message :class:`~.pincer.objects.message.message.Message`
263			The interaction message.
264			"""
265
266			if message.delete_after:
267			await sleep(message.delete_after)
268			await self.delete()
269
270			def __post_sent(self, message: Message):
271			"""Ensure the `__post_send_handler` method its future.
272
273			Parameters
274			----------
275			message :class:`~.pincer.objects.message.message.Message`
276			The interaction message.
277			"""
278			ensure_future(self.__post_send_handler(message))
279
280			async def reply(self, message: Message):
281			"""\|coro\|
282
283			Initial reply, only works if no ACK has been sent yet.
284
285			Parameters
286			----------
287			message :class:`~.pincer.objects.message.message.Message`
288			The response message!
289			"""
290			content_type, data = message.serialize()
291
292			await self._http.post(
293			f"interactions/{self.id}/{self.token}/callback",
294			{
295			"type": CallbackType.MESSAGE,
296			"data": data
297			},
298			content_type=content_type
299			)
300			self.__post_sent(message)
301
302			async def edit(self, message: Message) -> UserMessage:
303			"""\|coro\|
304
305			Edit an interaction. This is also the way to reply to
306			interactions whom have been acknowledged.
307
308			Parameters
309			----------
310			message :class:`~.pincer.objects.message.message.Message`
311			The new message!
312			"""
313			content_type, data = message.serialize()
314
315			resp = await self._http.patch(
316			f"webhooks/{self._client.bot.id}/{self.token}/messages/@original",
317			data,
318			content_type=content_type
319			)
320			self.__post_sent(message)
321			return UserMessage.from_dict(resp)
322
323			async def delete(self):
324			"""\|coro\|
325
326			Delete the interaction.
327			"""
328			await self._http.delete(
329			f"webhooks/{self._client.bot.id}/{self.token}/messages/@original"
330			)
331
332			async def __post_followup_send_handler(
333			self,
334			followup: UserMessage,
335			message: Message
336			):
337			"""Process a folloup after it was sent.
338
339			Parameters
340			----------
341			followup :class:`~.pincer.objects.message.user_message.UserMessage`
342			The followup message that is being post processed.
343			message :class:`~.pincer.objects.message.message.Message`
344			The followup message.
345			"""
346
347			if message.delete_after:
348			await sleep(message.delete_after)
349			await self.delete_followup(followup.id)
350
351			def __post_followup_sent(
352			self,
353			followup: UserMessage,
354			message: Message
355			):
356			"""Ensure the `__post_followup_send_handler` method its future.
357
358			Parameters
359			----------
360			followup :class:`~.pincer.objects.message.user_message.UserMessage`
361			The followup message that is being post processed.
362			message :class:`~.pincer.objects.message.message.Message`
363			The followup message.
364			"""
365			ensure_future(self.__post_followup_send_handler(followup, message))
366
367			async def followup(self, message: Message) -> UserMessage:
368			"""\|coro\|
369
370			Create a follow up message for the interaction.
371			This allows you to respond with multiple messages.
372
373			Parameters
374			----------
375			message :class:`~.pincer.objects.message.message.Message`
376			The message to sent.
377
378			Returns
379			-------
380			:class:`~.pincer.objects.message.user_message.UserMessage`
381			The message that has been sent.
382			"""
383			content_type, data = message.serialize()
384
385			resp = await self._http.post(
386			f"webhooks/{self._client.bot.id}/{self.token}",
387			data,
388			content_type=content_type
389			)
390			msg = UserMessage.from_dict(resp)
391			self.__post_followup_sent(msg, message)
392			return msg
393
394			async def edit_followup(self, message_id: int, message: Message) \
395			-> UserMessage:
396			"""\|coro\|
397
398			Edit a followup message.
399
400			Parameters
401			----------
402			message_id :class:`int`
403			The id of the original followup message.
404			message :class:`~.pincer.objects.message.message.Message`
405			The message to edit.
406
407			Returns
408			-------
409			:class:`~.pincer.objects.message.user_message.UserMessage`
410			The updated message object.
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			"""\|coro\|
425
426			Get a followup message by id.
427
428			Parameters
429			----------
430			message_id :class:`int`
431			The id of the original followup message that must be fetched.
432
433			Returns
434			-------
435			:class:`~.pincer.objects.message.user_message.UserMessage`
436			The fetched message object.
437			"""
438
439			resp = await self._http.get(
440			f"webhooks/{self._client.bot.id}/{self.token}/messages/{message_id}",
441			)
442			return UserMessage.from_dict(resp)
443
444			async def delete_followup(self, message_id: int):
445			"""\|coro\|
446
447			Remove a followup message by id.
448
449			Parameters
450			----------
451			message_id :class:`int`
452			The id of the followup message that must be deleted.
453			"""
454
455			await self._http.delete(
456			f"webhooks/{self._client.bot.id}/{self.token}/messages/{message_id}",
457			)
458

Pincer-org / Pincer

Pull Request — main (#158)

Interaction.__post_send_handler() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like