pincer.middleware.interaction_create.interaction_create_middleware() - Code Metrics - Inspection of ":sparkles: Implemented messagecontext" - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( c488a8...9a34ae )

unknown

created 2021-09-26 12:42 UTC

interaction_create_middleware() C

↳ Parent: pincer.middleware.interaction_create

Complexity

Conditions

Size

Total Lines	83
Code Lines	44

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	44
dl	0
loc	83
rs	6.4906
c	0
b	0
f	0
cc	9
nop	2

How to fix Long Method

# Copyright Pincer 2021-Present

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

import logging
from asyncio import sleep
from inspect import isasyncgenfunction
from typing import Union

from ..commands import ChatCommandHandler

from ..core.dispatch import GatewayDispatch
from ..exceptions import RateLimitError
from ..objects import Interaction, Embed, Message, InteractionFlags
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__)


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:
        defaults = {param: None for param in get_params(command.call)}
        params = {}

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

        kwargs = {**defaults, **params}

        if should_pass_cls(command.call):
            kwargs["self"] = self

        sig, params = get_signature_and_params(command.call)
        if should_pass_ctx(sig, params):
            kwargs[params[0]] = interaction.to_context()

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

            async for msg in message:

                msg = convert_message(self, msg)
                if started:
                    try:
                        await self.http.post(
                            f"webhooks/{interaction.application_id}"
                            f"/{interaction.token}",
                            msg.to_dict().get("data")
                        )
                    except RateLimitError as e:
                        _log.exception(

                            f"RateLimitError: {e}."

                            f" Retrying in {e.json.get('retry_after', 40)}"
                            f" seconds"
                        )

                        await sleep(e.json.get("retry_after", 40))

                        await self.http.post(
                            f"webhooks/{interaction.application_id}"
                            f"/{interaction.token}",
                            msg.to_dict().get("data")
                        )

                else:
                    started = True

                    await self.http.post(
                        f"interactions/{interaction.id}"
                        f"/{interaction.token}/callback",
                        msg.to_dict()
                    )
                await sleep(0.3)
        else:
            message = await command.call(**kwargs)
            message = convert_message(self, message)

            await self.http.post(
                f"interactions/{interaction.id}/{interaction.token}/callback",
                message.to_dict()
            )

    return "on_interaction_create", [interaction]


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


def export() -> Coro:

    return interaction_create_middleware


1			# Copyright Pincer 2021-Present
			0 ignored issues – show introduced 2021-09-13 13:25 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			import logging
5			from asyncio import sleep
6			from inspect import isasyncgenfunction
7			from typing import Union
8
9			from ..commands import ChatCommandHandler
			0 ignored issues – show introduced 2021-09-25 13:19 UTC by Report Bug Copy Issue Report Cannot import 'commands' due to syntax error 'invalid syntax (<unknown>, line 79)' Loading history...
10			from ..core.dispatch import GatewayDispatch
11			from ..exceptions import RateLimitError
12			from ..objects import Interaction, Embed, Message, InteractionFlags
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			async def interaction_create_middleware(self, payload: GatewayDispatch):
20			"""
21			Middleware for ``on_interaction``, which handles command
22			execution.
23
24			:param self:
25			The current client.
26
27			:param payload:
28			The data received from the interaction event.
29			"""
30
31			interaction: Interaction = Interaction.from_dict(
32			payload.data \| {"_client": self, "_http": self.http}
33			)
34
35			command = ChatCommandHandler.register.get(interaction.data.name)
36
37			if command:
38			defaults = {param: None for param in get_params(command.call)}
39			params = {}
40
41			if interaction.data.options is not MISSING:
42			params = {
43			opt.name: opt.value for opt in interaction.data.options
44			}
45
46			kwargs = {defaults, params}
47
48			if should_pass_cls(command.call):
49			kwargs["self"] = self
50
51			sig, params = get_signature_and_params(command.call)
52			if should_pass_ctx(sig, params):
53			kwargs[params[0]] = interaction.to_context()
54
55			if isasyncgenfunction(command.call):
56			message = command.call(**kwargs)
57			started = False
58
59			async for msg in message:
60
61			msg = convert_message(self, msg)
62			if started:
63			try:
64			await self.http.post(
65			f"webhooks/{interaction.application_id}"
66			f"/{interaction.token}",
67			msg.to_dict().get("data")
68			)
69			except RateLimitError as e:
70			_log.exception(
			0 ignored issues – show introduced 2021-09-19 19:51 UTC by Report Bug Copy Issue Report Use lazy % formatting in logging functions Loading history...
71			f"RateLimitError: {e}."
			0 ignored issues – show Bug introduced 2021-09-19 19:51 UTC by Report Bug Copy Issue Report The Instance of `RateLimitError` does not seem to have a member named `json`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
72			f" Retrying in {e.json.get('retry_after', 40)}"
73			f" seconds"
74			)
75
76			await sleep(e.json.get("retry_after", 40))
			0 ignored issues – show Bug introduced 2021-09-19 19:51 UTC by Report Bug Copy Issue Report The Instance of `RateLimitError` does not seem to have a member named `json`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
77			await self.http.post(
78			f"webhooks/{interaction.application_id}"
79			f"/{interaction.token}",
80			msg.to_dict().get("data")
81			)
82
83			else:
84			started = True
85
86			await self.http.post(
87			f"interactions/{interaction.id}"
88			f"/{interaction.token}/callback",
89			msg.to_dict()
90			)
91			await sleep(0.3)
92			else:
93			message = await command.call(**kwargs)
94			message = convert_message(self, message)
95
96			await self.http.post(
97			f"interactions/{interaction.id}/{interaction.token}/callback",
98			message.to_dict()
99			)
100
101			return "on_interaction_create", [interaction]
102
103
104			def convert_message(self, message: Union[Embed, Message, str]) -> Message:
105			"""Converts a message to a Message object"""
106			if isinstance(message, Embed):
107			message = Message(embeds=[message])
108			elif not isinstance(message, Message):
109			message = Message(message) if message else Message(
110			self.received_message,
111			flags=InteractionFlags.EPHEMERAL
112			)
113			return message
114
115
116			def export() -> Coro:
			0 ignored issues – show introduced 2021-09-14 08:25 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
117			return interaction_create_middleware
118

Pincer-org / Pincer

Push — main ( c488a8...9a34ae )

interaction_create_middleware() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like