pincer.middleware.interaction_create.export() - Code Metrics - Inspection of ":sparkles: Implemented rate limiting in the http c..." - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 6627c3...a229ba )

unknown

created 2021-09-27 16:02 UTC

pincer.middleware.interaction_create.export() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# Copyright Pincer 2021-Present
# Full MIT License can be found in `LICENSE` at the project root.

import logging
from inspect import isasyncgenfunction
from typing import Union, Dict, Any

from ..commands import ChatCommandHandler
from ..core.dispatch import GatewayDispatch
from ..objects import (
    Interaction, Embed, Message, InteractionFlags, MessageContext
)
from ..utils import MISSING, should_pass_cls, Coro, should_pass_ctx
from ..utils.extraction import get_params, get_signature_and_params

_log = logging.getLogger(__name__)


def convert_message(self, message: Union[Embed, Message, str]) -> Message:
    """Converts a message to a Message object"""
    if isinstance(message, Embed):
        message = Message(embeds=[message])
    elif not isinstance(message, Message):
        message = Message(message) if message else Message(
            self.received_message,
            flags=InteractionFlags.EPHEMERAL
        )
    return message


async def reply(self, interaction: Interaction, message):
    """
    Sends a reply to an interaction.

    :param self:
        The current client.

    :param interaction:
        The interaction from whom the reply is.

    :param message:
        The message to reply with.
    """
    await self.http.post(
        f"interactions/{interaction.id}/{interaction.token}/callback",
        message.to_dict()
    )


async def interaction_response_handler(
        self,
        command: Coro,
        context: MessageContext,
        interaction: Interaction,
        kwargs: Dict[str, Any]
):
    """
    Handle any coroutine as a command.

    :param self:
        The current client.

    :param command:
        The coroutine which will be seen as a command.

    :param context:
        The context of the command.

    :param interaction:
        The interaction which is linked to the command.

    :param kwargs:
        The arguments to be passed to the command.
    """
    # TODO: Make response thread based (eg a new thread per response handler)
    if should_pass_cls(command):
        kwargs["self"] = self

    sig, params = get_signature_and_params(command)
    if should_pass_ctx(sig, params):
        kwargs[params[0]] = context

    if isasyncgenfunction(command):
        message = command(**kwargs)
        started = False

        async for msg in message:
            msg = convert_message(self, msg)

            if started:
                await self.http.post(
                    f"webhooks/{interaction.application_id}"
                    f"/{interaction.token}",
                    msg.to_dict().get("data")
                )
            else:
                started = True
                await reply(self, interaction, msg)
    else:
        message = await command(**kwargs)
        await reply(self, interaction, convert_message(self, message))


async def interaction_handler(
        self,
        interaction: Interaction,
        context: MessageContext,
        command: Coro
):
    """
    Processes an interaction.

    :param self:
        The current client.

    :param interaction:
        The interaction which is linked to the command.

    :param context:
        The context of the command.

    :param command:
        The coroutine which will be seen as a command.
    """
    self.throttler.handle(context)

    defaults = {param: None for param in get_params(command)}
    params = {}

    if interaction.data.options is not MISSING:
        params = {
            opt.name: opt.value for opt in interaction.data.options
        }

    kwargs = {**defaults, **params}

    await interaction_response_handler(self, command, context, interaction,
                                       kwargs)


async def interaction_create_middleware(self, payload: GatewayDispatch):
    """
    Middleware for ``on_interaction``, which handles command
    execution.

    :param self:
        The current client.

    :param payload:
        The data received from the interaction event.
    """

    interaction: Interaction = Interaction.from_dict(
        {**payload.data, "_client": self, "_http": self.http}
    )
    command = ChatCommandHandler.register.get(interaction.data.name)

    if command:
        context = interaction.convert_to_message_context(command)

        try:
            await interaction_handler(self, interaction, context,
                                      command.call)
        except Exception as e:
            if coro := self.get_event_coro("on_command_error"):

                params = get_signature_and_params(coro)[1]

                # Check if a context or error var has been passed.
                if 0 < len(params) < 3:
                    await interaction_response_handler(
                        self,
                        coro,
                        context,
                        interaction,
                        # Always take the error parameter its name.
                        {params[(len(params) - 1) or 0]: e}
                    )
                else:
                    raise e
            else:
                raise e

    return "on_interaction_create", [interaction]


def export() -> Coro:
    return interaction_create_middleware


1			# Copyright Pincer 2021-Present
2			# Full MIT License can be found in `LICENSE` at the project root.
3
4			import logging
5			from inspect import isasyncgenfunction
6			from typing import Union, Dict, Any
7
8			from ..commands import ChatCommandHandler
9			from ..core.dispatch import GatewayDispatch
10			from ..objects import (
11			Interaction, Embed, Message, InteractionFlags, MessageContext
12			)
13			from ..utils import MISSING, should_pass_cls, Coro, should_pass_ctx
14			from ..utils.extraction import get_params, get_signature_and_params
15
16			_log = logging.getLogger(__name__)
17
18
19			def convert_message(self, message: Union[Embed, Message, str]) -> Message:
20			"""Converts a message to a Message object"""
21			if isinstance(message, Embed):
22			message = Message(embeds=[message])
23			elif not isinstance(message, Message):
24			message = Message(message) if message else Message(
25			self.received_message,
26			flags=InteractionFlags.EPHEMERAL
27			)
28			return message
29
30
31			async def reply(self, interaction: Interaction, message):
32			"""
33			Sends a reply to an interaction.
34
35			:param self:
36			The current client.
37
38			:param interaction:
39			The interaction from whom the reply is.
40
41			:param message:
42			The message to reply with.
43			"""
44			await self.http.post(
45			f"interactions/{interaction.id}/{interaction.token}/callback",
46			message.to_dict()
47			)
48
49
50			async def interaction_response_handler(
51			self,
52			command: Coro,
53			context: MessageContext,
54			interaction: Interaction,
55			kwargs: Dict[str, Any]
56			):
57			"""
58			Handle any coroutine as a command.
59
60			:param self:
61			The current client.
62
63			:param command:
64			The coroutine which will be seen as a command.
65
66			:param context:
67			The context of the command.
68
69			:param interaction:
70			The interaction which is linked to the command.
71
72			:param kwargs:
73			The arguments to be passed to the command.
74			"""
75			# TODO: Make response thread based (eg a new thread per response handler)
76			if should_pass_cls(command):
77			kwargs["self"] = self
78
79			sig, params = get_signature_and_params(command)
80			if should_pass_ctx(sig, params):
81			kwargs[params[0]] = context
82
83			if isasyncgenfunction(command):
84			message = command(**kwargs)
85			started = False
86
87			async for msg in message:
88			msg = convert_message(self, msg)
89
90			if started:
91			await self.http.post(
92			f"webhooks/{interaction.application_id}"
93			f"/{interaction.token}",
94			msg.to_dict().get("data")
95			)
96			else:
97			started = True
98			await reply(self, interaction, msg)
99			else:
100			message = await command(**kwargs)
101			await reply(self, interaction, convert_message(self, message))
102
103
104			async def interaction_handler(
105			self,
106			interaction: Interaction,
107			context: MessageContext,
108			command: Coro
109			):
110			"""
111			Processes an interaction.
112
113			:param self:
114			The current client.
115
116			:param interaction:
117			The interaction which is linked to the command.
118
119			:param context:
120			The context of the command.
121
122			:param command:
123			The coroutine which will be seen as a command.
124			"""
125			self.throttler.handle(context)
126
127			defaults = {param: None for param in get_params(command)}
128			params = {}
129
130			if interaction.data.options is not MISSING:
131			params = {
132			opt.name: opt.value for opt in interaction.data.options
133			}
134
135			kwargs = {defaults, params}
136
137			await interaction_response_handler(self, command, context, interaction,
138			kwargs)
139
140
141			async def interaction_create_middleware(self, payload: GatewayDispatch):
142			"""
143			Middleware for ``on_interaction``, which handles command
144			execution.
145
146			:param self:
147			The current client.
148
149			:param payload:
150			The data received from the interaction event.
151			"""
152
153			interaction: Interaction = Interaction.from_dict(
154			{**payload.data, "_client": self, "_http": self.http}
155			)
156			command = ChatCommandHandler.register.get(interaction.data.name)
157
158			if command:
159			context = interaction.convert_to_message_context(command)
160
161			try:
162			await interaction_handler(self, interaction, context,
163			command.call)
164			except Exception as e:
165			if coro := self.get_event_coro("on_command_error"):
			0 ignored issues – show introduced 2021-09-27 16:04 UTC by Report Bug Copy Issue Report invalid syntax (<unknown>, line 165) Loading history...
166			params = get_signature_and_params(coro)[1]
167
168			# Check if a context or error var has been passed.
169			if 0 < len(params) < 3:
170			await interaction_response_handler(
171			self,
172			coro,
173			context,
174			interaction,
175			# Always take the error parameter its name.
176			{params[(len(params) - 1) or 0]: e}
177			)
178			else:
179			raise e
180			else:
181			raise e
182
183			return "on_interaction_create", [interaction]
184
185
186			def export() -> Coro:
187			return interaction_create_middleware
188

Pincer-org / Pincer

Push — main ( 6627c3...a229ba )

pincer.middleware.interaction_create.export() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like