pincer.client.Client.__init__() - Code Metrics - Inspection of ":recycle: Refactored some things and typo fixes." - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 6933d2...87b586 )

unknown

created 2021-08-31 20:43 UTC

pincer.client.Client.init() B

↳ Parent: pincer.client

Complexity

Conditions

Size

Total Lines	102
Code Lines	96

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	96
dl	0
loc	102
rs	7.1309
c	0
b	0
f	0
cc	1
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.
import logging
from asyncio import iscoroutinefunction
from typing import Optional, TypeVar, Callable, Coroutine, Any, Union

from pincer import __package__

from pincer._config import GatewayConfig
from pincer.core.dispatch import GatewayDispatch
from pincer.core.gateway import Dispatcher
from pincer.core.http import HTTPClient
from pincer.exceptions import InvalidEventName
from pincer.objects.user import User

_log = logging.getLogger(__package__)

Coro = TypeVar('Coro', bound=Callable[..., Coroutine[Any, Any, Any]])


class Client(Dispatcher):
    """
    The main instance which the user will interact with.
    """

    def __init__(self, token: str):
        # TODO: Write docs

        # TODO: Implement intents

        super().__init__(
            token,
            handlers={
                0: self.event_handler
            }
        )

        self.http = HTTPClient(token, GatewayConfig.version)
        self.bot: Optional[User] = None

        self.__events = {
            "ready": self.__on_ready,
            "on_ready": None,
            "channel_create": "on_channel_create",
            "on_channel_create": None,
            "channel_update": "on_channel_update",
            "on_channel_update": None,
            "channel_delete": "on_channel_delete",
            "on_channel_delete": None,
            "channel_pin_update": "on_channel_pin_update",
            "on_channel_pin_update": None,
            "thread_create": "on_thread_create",
            "on_thread_create": None,
            "thread_update": "on_thread_update",
            "on_thread_update": None,
            "thread_delete": "on_thread_delete",
            "on_thread_delete": None,
            "thread_member_update": "on_thread_member_update",
            "on_thread_member_update": None,
            "thread_members_update": "on_thread_members_update",
            "on_thread_members_update": None,
            "guild_create": "on_guild_create",
            "on_guild_create": None,
            "guild_update": "on_guild_update",
            "on_guild_update": None,
            "guild_delete": "on_guild_delete",
            "on_guild_delete": None,
            "guild_ban_add": "on_guild_ban_add",
            "on_guild_ban_add": None,
            "guild_ban_remove": "on_guild_ban_remove",
            "on_guild_ban_remove": None,
            "guild_emoji_update": "on_guild_emoji_update",
            "on_guild_emoji_update": None,
            "guild_stickers_update": "on_guild_stickers_update",
            "on_guild_stickers_update": None,
            "guild_integrations_update": "on_guild_integrations_update",
            "on_guild_integrations_update": None,
            "guild_member_add": "on_guild_member_add",
            "on_guild_member_add": None,
            "guild_member_remove": "on_guild_member_remove",
            "on_guild_member_remove": None,
            "guild_members_chunk": "on_guild_members_chunk",
            "on_guild_members_chunk": None,
            "guild_role_create": "on_guild_role_create",
            "on_guild_role_create": None,
            "guild_role_update": "on_guild_role_update",
            "on_guild_role_update": None,
            "guild_role_delete": "on_guild_role_delete",
            "on_guild_role_delete": None,
            "integration_create": "on_integration_create",
            "on_integration_create": None,
            "integration_update": "on_integration_update",
            "on_integration_update": None,
            "integration_delete": "on_integration_delete",
            "on_integration_delete": None,
            "invite_create": "on_invite_create",
            "on_invite_create": None,
            "messages_create": "on_messages_create",
            "on_messages_create": None,
            "message_update": "on_message_update",
            "on_message_update": None,
            "message_delete": "on_message_delete",
            "on_message_delete": None,
            "message_delete_bulk": "on_message_delete_bulk",
            "on_message_delete_bulk": None,
            "message_reaction_add": "on_message_reaction_add",
            "on_message_reaction_add": None,
            "message_reaction_remove": "on_message_reaction_remove",
            "on_message_reaction_remove": None,
            "message_reaction_remove_all": "on_message_reaction_remove_all",
            "on_message_reaction_remove_all": None,
            "message_reaction_remove_emoji": "on_message_reaction_remove_emoji",
            "on_message_reaction_remove_emoji": None,
            "presence_update": "on_presence_update",
            "on_presence_update": None,
            "stage_instance_create": "on_stage_instance_create",
            "on_stage_instance_create": None,
            "stage_instance_update": "on_stage_instance_update",
            "on_stage_instance_update": None,
            "stage_instance_delete": "on_stage_instance_delete",
            "on_stage_instance_delete": None,
            "typing_start": "on_typing_start",
            "on_typing_start": None,
            "voice_state_update": "on_voice_state_update",
            "on_voice_state_update": None,
            "voice_server_update": "on_voice_server_update",
            "on_voice_server_update": None,
            "webhooks_update": "on_webhooks_update",
            "on_webhooks_update": None,
        }

    def event(self, coroutine: Coro):

        if not iscoroutinefunction(coroutine):
            raise TypeError("Any event which is registered must be a coroutine "
                            "function")

        name: str = coroutine.__name__.lower()

        if not name.startswith("on_"):
            raise InvalidEventName(
                f"The event `{name}` its name must start with `on_`"
            )

        if self.__events.get(name) is not None:
            raise InvalidEventName(
                f"The event `{name}` has already been registered or is not "
                f"a event name."
            )

        self.__events[name] = coroutine
        return coroutine

    async def event_handler(self, _, payload: GatewayDispatch):

        middleware: Optional[Union[Coro, str]] = self.__events.get(
            payload.event_name.lower()
        )

        if iscoroutinefunction(middleware):
            final_call, params = await middleware(payload)
        else:
            final_call, params = middleware, dict()

        final_call: str = final_call
        params: dict = params

        final_call_routine: Optional[Coro] = self.__events.get(final_call)

        if iscoroutinefunction(final_call_routine):
            await final_call_routine(**params)

    async def __on_ready(self, payload: GatewayDispatch):
        self.bot = User.from_dict(payload.data.get("user"))
        return "on_ready", dict()


Bot = Client


1			# -- coding: utf-8 --
			0 ignored issues – show introduced 2021-08-31 17:57 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			import logging
25			from asyncio import iscoroutinefunction
26			from typing import Optional, TypeVar, Callable, Coroutine, Any, Union
27
28			from pincer import __package__
			1 ignored issue – show Bug Best Practice introduced 2021-08-31 20:44 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `__package__`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
29			from pincer._config import GatewayConfig
30			from pincer.core.dispatch import GatewayDispatch
31			from pincer.core.gateway import Dispatcher
32			from pincer.core.http import HTTPClient
33			from pincer.exceptions import InvalidEventName
34			from pincer.objects.user import User
35
36			_log = logging.getLogger(__package__)
37
38			Coro = TypeVar('Coro', bound=Callable[..., Coroutine[Any, Any, Any]])
39
40
41			class Client(Dispatcher):
42			"""
43			The main instance which the user will interact with.
44			"""
45
46			def __init__(self, token: str):
47			# TODO: Write docs
			0 ignored issues – show Coding Style introduced 2021-08-31 17:57 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
48			# TODO: Implement intents
			0 ignored issues – show Coding Style introduced 2021-08-31 17:57 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
49			super().__init__(
50			token,
51			handlers={
52			0: self.event_handler
53			}
54			)
55
56			self.http = HTTPClient(token, GatewayConfig.version)
57			self.bot: Optional[User] = None
58
59			self.__events = {
60			"ready": self.__on_ready,
61			"on_ready": None,
62			"channel_create": "on_channel_create",
63			"on_channel_create": None,
64			"channel_update": "on_channel_update",
65			"on_channel_update": None,
66			"channel_delete": "on_channel_delete",
67			"on_channel_delete": None,
68			"channel_pin_update": "on_channel_pin_update",
69			"on_channel_pin_update": None,
70			"thread_create": "on_thread_create",
71			"on_thread_create": None,
72			"thread_update": "on_thread_update",
73			"on_thread_update": None,
74			"thread_delete": "on_thread_delete",
75			"on_thread_delete": None,
76			"thread_member_update": "on_thread_member_update",
77			"on_thread_member_update": None,
78			"thread_members_update": "on_thread_members_update",
79			"on_thread_members_update": None,
80			"guild_create": "on_guild_create",
81			"on_guild_create": None,
82			"guild_update": "on_guild_update",
83			"on_guild_update": None,
84			"guild_delete": "on_guild_delete",
85			"on_guild_delete": None,
86			"guild_ban_add": "on_guild_ban_add",
87			"on_guild_ban_add": None,
88			"guild_ban_remove": "on_guild_ban_remove",
89			"on_guild_ban_remove": None,
90			"guild_emoji_update": "on_guild_emoji_update",
91			"on_guild_emoji_update": None,
92			"guild_stickers_update": "on_guild_stickers_update",
93			"on_guild_stickers_update": None,
94			"guild_integrations_update": "on_guild_integrations_update",
95			"on_guild_integrations_update": None,
96			"guild_member_add": "on_guild_member_add",
97			"on_guild_member_add": None,
98			"guild_member_remove": "on_guild_member_remove",
99			"on_guild_member_remove": None,
100			"guild_members_chunk": "on_guild_members_chunk",
101			"on_guild_members_chunk": None,
102			"guild_role_create": "on_guild_role_create",
103			"on_guild_role_create": None,
104			"guild_role_update": "on_guild_role_update",
105			"on_guild_role_update": None,
106			"guild_role_delete": "on_guild_role_delete",
107			"on_guild_role_delete": None,
108			"integration_create": "on_integration_create",
109			"on_integration_create": None,
110			"integration_update": "on_integration_update",
111			"on_integration_update": None,
112			"integration_delete": "on_integration_delete",
113			"on_integration_delete": None,
114			"invite_create": "on_invite_create",
115			"on_invite_create": None,
116			"messages_create": "on_messages_create",
117			"on_messages_create": None,
118			"message_update": "on_message_update",
119			"on_message_update": None,
120			"message_delete": "on_message_delete",
121			"on_message_delete": None,
122			"message_delete_bulk": "on_message_delete_bulk",
123			"on_message_delete_bulk": None,
124			"message_reaction_add": "on_message_reaction_add",
125			"on_message_reaction_add": None,
126			"message_reaction_remove": "on_message_reaction_remove",
127			"on_message_reaction_remove": None,
128			"message_reaction_remove_all": "on_message_reaction_remove_all",
129			"on_message_reaction_remove_all": None,
130			"message_reaction_remove_emoji": "on_message_reaction_remove_emoji",
131			"on_message_reaction_remove_emoji": None,
132			"presence_update": "on_presence_update",
133			"on_presence_update": None,
134			"stage_instance_create": "on_stage_instance_create",
135			"on_stage_instance_create": None,
136			"stage_instance_update": "on_stage_instance_update",
137			"on_stage_instance_update": None,
138			"stage_instance_delete": "on_stage_instance_delete",
139			"on_stage_instance_delete": None,
140			"typing_start": "on_typing_start",
141			"on_typing_start": None,
142			"voice_state_update": "on_voice_state_update",
143			"on_voice_state_update": None,
144			"voice_server_update": "on_voice_server_update",
145			"on_voice_server_update": None,
146			"webhooks_update": "on_webhooks_update",
147			"on_webhooks_update": None,
148			}
149
150			def event(self, coroutine: Coro):
			0 ignored issues – show introduced 2021-08-31 17:57 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
151			if not iscoroutinefunction(coroutine):
152			raise TypeError("Any event which is registered must be a coroutine "
153			"function")
154
155			name: str = coroutine.__name__.lower()
156
157			if not name.startswith("on_"):
158			raise InvalidEventName(
159			f"The event `{name}` its name must start with `on_`"
160			)
161
162			if self.__events.get(name) is not None:
163			raise InvalidEventName(
164			f"The event `{name}` has already been registered or is not "
165			f"a event name."
166			)
167
168			self.__events[name] = coroutine
169			return coroutine
170
171			async def event_handler(self, _, payload: GatewayDispatch):
			0 ignored issues – show introduced 2021-08-31 20:44 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
172			middleware: Optional[Union[Coro, str]] = self.__events.get(
173			payload.event_name.lower()
174			)
175
176			if iscoroutinefunction(middleware):
177			final_call, params = await middleware(payload)
178			else:
179			final_call, params = middleware, dict()
180
181			final_call: str = final_call
182			params: dict = params
183
184			final_call_routine: Optional[Coro] = self.__events.get(final_call)
185
186			if iscoroutinefunction(final_call_routine):
187			await final_call_routine(**params)
188
189			async def __on_ready(self, payload: GatewayDispatch):
190			self.bot = User.from_dict(payload.data.get("user"))
191			return "on_ready", dict()
192
193
194			Bot = Client
195

Pincer-org / Pincer

Push — main ( 6933d2...87b586 )

pincer.client.Client.__init__() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

pincer.client.Client.init() B