pincer.middleware.interaction_create.interaction_create_middleware() - Code Metrics - Inspection of "Yieldable commands" - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#70)

unknown

created 2021-09-19 19:48 UTC

interaction_create_middleware() B

↳ Parent: pincer.middleware.interaction_create

Complexity

Conditions

Size

Total Lines	68
Code Lines	40

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	40
dl	0
loc	68
rs	7.0533
c	0
b	0
f	0
cc	8
nop	2

How to fix Long Method

# -*- coding: utf-8 -*-

# MIT License
#
# Copyright (c) 2021 Pincer
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from typing import Union
from asyncio import sleep
from inspect import isasyncgenfunction
import logging

from pincer.commands import ChatCommandHandler

from pincer.core.dispatch import GatewayDispatch
from pincer.objects import Interaction, Embed, Message, InteractionFlags
from pincer.exceptions import RateLimitError
from pincer.utils import MISSING, should_pass_cls, Coro
from pincer.utils.extraction import get_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)
    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

        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}/{interaction.token}",
                            msg.to_dict().get("data")
                        )
                    except RateLimitError as e:
                        _log.exception(

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

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

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

                else:
                    started = True

                    await self.http.post(
                        f"interactions/{interaction.id}/{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			# -- coding: utf-8 --
			0 ignored issues – show introduced 2021-09-13 13:25 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			# MIT License
3			#
4			# Copyright (c) 2021 Pincer
5			#
6			# Permission is hereby granted, free of charge, to any person obtaining
7			# a copy of this software and associated documentation files
8			# (the "Software"), to deal in the Software without restriction,
9			# including without limitation the rights to use, copy, modify, merge,
10			# publish, distribute, sublicense, and/or sell copies of the Software,
11			# and to permit persons to whom the Software is furnished to do so,
12			# subject to the following conditions:
13			#
14			# The above copyright notice and this permission notice shall be
15			# included in all copies or substantial portions of the Software.
16			#
17			# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
18			# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
19			# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
20			# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
21			# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
22			# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
23			# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
24			from typing import Union
25			from asyncio import sleep
26			from inspect import isasyncgenfunction
27			import logging
28
29			from pincer.commands import ChatCommandHandler
			0 ignored issues – show Bug introduced 2021-09-13 13:25 UTC by Report Bug Copy Issue Report The name `commands` does not seem to exist in module `pincer`. Loading history... introduced 2021-09-13 13:25 UTC by Report Bug Copy Issue Report Cannot import 'pincer.commands' due to syntax error 'invalid syntax (<unknown>, line 89)' Loading history...
30			from pincer.core.dispatch import GatewayDispatch
31			from pincer.objects import Interaction, Embed, Message, InteractionFlags
32			from pincer.exceptions import RateLimitError
33			from pincer.utils import MISSING, should_pass_cls, Coro
34			from pincer.utils.extraction import get_params
35
36			_log = logging.getLogger(__name__)
37
38
39			async def interaction_create_middleware(self, payload: GatewayDispatch):
40			"""
41			Middleware for ``on_interaction``, which handles command
42			execution.
43
44			:param self:
45			The current client.
46
47			:param payload:
48			The data received from the interaction event.
49			"""
50			interaction: Interaction = Interaction.from_dict(payload.data)
51			command = ChatCommandHandler.register.get(interaction.data.name)
52
53			if command:
54			defaults = {param: None for param in get_params(command.call)}
55			params = {}
56
57			if interaction.data.options is not MISSING:
58			params = {
59			opt.name: opt.value for opt in interaction.data.options
60			}
61
62			kwargs = {defaults, params}
63
64			if should_pass_cls(command.call):
65			kwargs["self"] = self
66
67			if isasyncgenfunction(command.call):
68			message = command.call(**kwargs)
69			started = False
70
71			async for msg in message:
72
73			msg = convert_message(self, msg)
74			if started:
75			try:
76			await self.http.post(
77			f"webhooks/{interaction.application_id}/{interaction.token}",
78			msg.to_dict().get("data")
79			)
80			except RateLimitError as e:
81			_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...
82			f"RateLimitError: {e}. Retrying in {e.json.get('retry_after', 40)} seconds")
			0 ignored issues – show Coding Style introduced 2021-09-19 19:51 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (104/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history... 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...
83			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...
84			await self.http.post(
85			f"webhooks/{interaction.application_id}/{interaction.token}",
86			msg.to_dict().get("data")
87			)
88
89			else:
90			started = True
91
92			await self.http.post(
93			f"interactions/{interaction.id}/{interaction.token}/callback",
94			msg.to_dict()
95			)
96			await sleep(0.3)
97			else:
98			message = await command.call(**kwargs)
99			message = convert_message(self, message)
100
101			await self.http.post(
102			f"interactions/{interaction.id}/{interaction.token}/callback",
103			message.to_dict()
104			)
105
106			return "on_interaction_create", [interaction]
107
108
109			def convert_message(self, message: Union[Embed, Message, str]) -> Message:
110			"""Converts a message to a Message object"""
111			if isinstance(message, Embed):
112			message = Message(embeds=[message])
113			elif not isinstance(message, Message):
114			message = Message(message) if message else Message(
115			self.received_message,
116			flags=InteractionFlags.EPHEMERAL
117			)
118			return message
119
120
121			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...
122			return interaction_create_middleware
123

Pincer-org / Pincer

Pull Request — main (#70)

interaction_create_middleware() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like