pincer.objects.app.interactions.Interaction.__post_sent() - 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 15:43 UTC

Interaction.__post_sent() A

↳ Parent: pincer.objects.app.interactions

Complexity

Conditions

Size

Total Lines	8
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	8
rs	10
c	0
b	0
f	0
cc	1
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):
        """
        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):
        """
        Initial reply, only works if no ACK has been sent yet.

        :param 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:
        """
        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 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			"""
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 2021-09-26 12:43 UTC by Report Bug Copy Issue Report 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

Pincer-org / Pincer

Pull Request — main (#158)

Interaction.__post_sent() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like