pincer.client.Client.event() - Code Metrics - Inspection of ":sparkles: Implemented basic commands (normal argu..." - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( f6a351...3d8f6e )

unknown

created 2021-09-09 12:22 UTC

pincer.client.Client.event() A

↳ Parent: pincer.client

Complexity

Conditions

Size

Total Lines	74
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	14
dl	0
loc	74
rs	9.7
c	0
b	0
f	0
cc	4
nop	1

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 __future__ import annotations

import logging
from asyncio import iscoroutinefunction
from typing import Optional, Any, Union, Dict, Tuple, List

from . import __package__
from ._config import events
from .commands import ChatCommandHandler

from .core.dispatch import GatewayDispatch
from .core.gateway import Dispatcher
from .core.http import HTTPClient
from .exceptions import InvalidEventName
from .objects import User
from .objects.interactions import Interaction
from .utils import get_index, should_pass_cls, Coro

_log = logging.getLogger(__package__)

MiddlewareType = Optional[Union[Coro, Tuple[str, List[Any], Dict[str, Any]]]]

_events: Dict[str, Optional[Union[str, Coro]]] = {}

for event in events:
    event_final_executor = f"on_{event}"

    # Event middleware for the library.
    # Function argument is a payload (GatewayDispatch).

    # The function must return a string which
    # contains the main event key.

    # As second value a list with arguments,
    # and thee third value value must be a dictionary.
    # The last two are passed on as *args and **kwargs.

    # NOTE: These return values must be passed as a tuple!
    _events[event] = event_final_executor

    # The registered event by the client. Do not manually overwrite.
    _events[event_final_executor] = None


def middleware(call: str, *, override: bool = False):
    """
    Middleware are methods which can be registered with this decorator.
    These methods are invoked before any ``on_`` event.
    As the ``on_`` event is the final call.

    A default call exists for all events, but some might already be in
    use by the library.

    If you know what you are doing, you can override these default
    middleware methods by passing the override parameter.

    The method to which this decorator is registered must be a coroutine,
    and it must return a tuple with the following format:

    .. code-block:: python

        tuple(
            key for next middleware or final event [str],
            args for next middleware/event which will be passed as *args
                [list(Any)],
            kwargs for next middleware/event which will be passed as
                **kwargs [dict(Any)]
        )

    Two parameters are passed to the middleware. The first parameter is
    the current socket connection with and the second one is the payload
    parameter which is of type :class:`~.core.dispatch.GatewayDispatch`.
    This contains the response from the discord API.

    :Implementation example:

    .. code-block:: pycon

        >>> @middleware("ready", override=True)
        >>> async def custom_ready(_, payload: GatewayDispatch):
        >>>     return "on_ready", [User.from_dict(payload.data.get("user"))]

        >>> @Client.event
        >>> async def on_ready(bot: User):
        >>>     print(f"Signed in as {bot}")


    :param call:
        The call where the method should be registered.

    Keyword Arguments:

    :param override:
        Setting this to True will allow you to override existing
        middleware. Usage of this is discouraged, but can help you out
        of some situations.
    """

    def decorator(func: Coro):
        if override:
            _log.warning(
                "Middleware overriding has been enabled for `%s`."
                " This might cause unexpected behavior.", call
            )

        if not override and callable(_events.get(call)):
            raise RuntimeError(
                f"Middleware event with call `{call}` has "
                "already been registered"
            )

        async def wrapper(cls, payload: GatewayDispatch):
            _log.debug("`%s` middleware has been invoked", call)

            return await (
                func(cls, payload)
                if should_pass_cls(func)
                else await func(payload)
            )

        _events[call] = wrapper
        return wrapper

    return decorator


class Client(Dispatcher):

    def __init__(self, token: str):
        """
        The client is the main instance which is between the programmer
            and the discord API.

        This client represents your bot.

        :param token:
            The secret bot token which can be found in
            `<https://discord.com/developers/applications/\<bot_id\>/bot>`_
        """
        # TODO: Implement intents

        super().__init__(
            token,
            handlers={
                # Use this event handler for opcode 0.
                0: self.event_handler
            }
        )

        self.bot: Optional[User] = None
        self.__token = token

    @property
    def http(self):
        """
        Returns a http client with the current client its
        authentication credentials.

        :Usage example:

        .. code-block:: pycon

            >>> async with self.http as client:
            >>>     await client.post(
            ...         '<endpoint>',
            ...         {
            ...             "foo": "bar",
            ...             "bar": "baz",
            ...             "baz": "foo"
            ...         }
            ...    )

        """
        return HTTPClient(self.__token)

    @staticmethod
    def event(coroutine: Coro):
        """
        Register a Discord gateway event listener. This event will get
        called when the client receives a new event update from Discord
        which matches the event name.

        The event name gets pulled from your method name, and this must
        start with ``on_``. This forces you to write clean and consistent
        code.

        This decorator can be used in and out of a class, and all
        event methods must be coroutines. *(async)*

        :Example usage:

        .. code-block:: pycon

            >>> # Function based
            >>> from pincer import Client
            >>>
            >>> client = Client("token")
            >>>
            >>> @client.event
            >>> async def on_ready():
            ...     print(f"Signed in as {client.bot}")
            >>>
            >>> if __name__ == "__main__":
            ...     client.run()

        .. code-block :: pycon

            >>> # Class based
            >>> from pincer import Client
            >>>
            >>> class BotClient(Client):
            ...     @Client.event
            ...     async def on_ready(self):
            ...         print(f"Signed in as {self.bot}")
            >>>
            >>> if __name__ == "__main__":
            ...     BotClient("token").run()


        :param coroutine: # TODO: add info

        :raises TypeError:
            If the method is not a coroutine.

        :raises InvalidEventName:
            If the event name does not start with ``on_``, has already
            been registered or is not a valid event name.
        """

        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 _events.get(name) is not None:
            raise InvalidEventName(
                f"The event `{name}` has already been registered or is not "
                f"a valid event name."
            )

        _events[name] = coroutine
        return coroutine

    async def handle_middleware(
            self,
            payload: GatewayDispatch,
            key: str,
            *args,
            **kwargs
    ) -> Tuple[Optional[Coro], List[Any], Dict[str, Any]]:
        """
        Handles all middleware recursively. Stops when it has found an
        event name which starts with ``on_``.

        :param payload:
            The original payload for the event.

        :param key:
            The index of the middleware in ``_events``.

        :param \*args:
            The arguments which will be passed to the middleware.

        :param \*\*kwargs:
            The named arguments which will be passed to the middleware.

        :return:
            A tuple where the first element is the final executor
            (so the event) its index in ``_events``. The second and third
            element are the ``*args`` and ``**kwargs`` for the event.
        """
        ware: MiddlewareType = _events.get(key)
        next_call, arguments, params = ware, [], {}

        if iscoroutinefunction(ware):
            extractable = await ware(self, payload, *args, **kwargs)

            if not isinstance(extractable, tuple):
                raise RuntimeError(
                    f"Return type from `{key}` middleware must be tuple. "
                )

            next_call = get_index(extractable, 0, "")
            arguments = get_index(extractable, 1, [])
            params = get_index(extractable, 2, {})

        if next_call is None:
            raise RuntimeError(f"Middleware `{key}` has not been registered.")

        return (
            (next_call, arguments, params)
            if next_call.startswith("on_")
            else await self.handle_middleware(
                payload, next_call, *arguments, **params
            )
        )

    async def event_handler(self, _, payload: GatewayDispatch):
        """
        Handles all payload events with opcode 0.

        :param _:
            Socket param, but this isn't required for this handler. So
            its just a filler parameter, doesn't matter what is passed.

        :param payload:
            The payload sent from the Discord gateway, this contains the
            required data for the client to know what event it is and
            what specifically happened.
        """
        event_name = payload.event_name.lower()

        key, args, kwargs = await self.handle_middleware(payload, event_name)

        call = _events.get(key)

        if iscoroutinefunction(call):
            if should_pass_cls(call):
                await call(self, *args, **kwargs)
            else:
                await call(*args, **kwargs)

    @middleware("ready")
    async def on_ready_middleware(self, payload: GatewayDispatch):
        """
        Middleware for ``on_ready`` event.

        :param payload:
            The data received from the ready event.
        """
        self.bot = User.from_dict(payload.data.get("user"))
        return "on_ready",

    @middleware("interaction_create")
    async def on_interaction_middleware(self, payload: GatewayDispatch):
        """
        Middleware for ``on_interaction``, which handles command
        execution.

        :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:
            kwargs = {opt.name: opt.value for opt in interaction.data.options}

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

            res = await command.call(**kwargs)

            if res:
                async with self.http as http:
                    await http.post(f"interactions/{interaction.id}/{interaction.token}/callback", {
                        "type": 4,
                        "data": {
                            "content": str(res)
                        }
                    })

        return "on_interaction_create", [interaction]


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
25			from __future__ import annotations
26
27			import logging
28			from asyncio import iscoroutinefunction
29			from typing import Optional, Any, Union, Dict, Tuple, List
30
31			from . import __package__
32			from ._config import events
33			from .commands import ChatCommandHandler
			0 ignored issues – show introduced 2021-09-09 12:24 UTC by Report Bug Copy Issue Report Cannot import 'commands' due to syntax error 'invalid syntax (<unknown>, line 80)' Loading history...
34			from .core.dispatch import GatewayDispatch
35			from .core.gateway import Dispatcher
36			from .core.http import HTTPClient
37			from .exceptions import InvalidEventName
38			from .objects import User
39			from .objects.interactions import Interaction
40			from .utils import get_index, should_pass_cls, Coro
41
42			_log = logging.getLogger(__package__)
43
44			MiddlewareType = Optional[Union[Coro, Tuple[str, List[Any], Dict[str, Any]]]]
45
46			_events: Dict[str, Optional[Union[str, Coro]]] = {}
47
48			for event in events:
49			event_final_executor = f"on_{event}"
50
51			# Event middleware for the library.
52			# Function argument is a payload (GatewayDispatch).
53
54			# The function must return a string which
55			# contains the main event key.
56
57			# As second value a list with arguments,
58			# and thee third value value must be a dictionary.
59			# The last two are passed on as args and *kwargs.
60
61			# NOTE: These return values must be passed as a tuple!
62			_events[event] = event_final_executor
63
64			# The registered event by the client. Do not manually overwrite.
65			_events[event_final_executor] = None
66
67
68			def middleware(call: str, *, override: bool = False):
69			"""
70			Middleware are methods which can be registered with this decorator.
71			These methods are invoked before any ``on_`` event.
72			As the ``on_`` event is the final call.
73
74			A default call exists for all events, but some might already be in
75			use by the library.
76
77			If you know what you are doing, you can override these default
78			middleware methods by passing the override parameter.
79
80			The method to which this decorator is registered must be a coroutine,
81			and it must return a tuple with the following format:
82
83			.. code-block:: python
84
85			tuple(
86			key for next middleware or final event [str],
87			args for next middleware/event which will be passed as *args
88			[list(Any)],
89			kwargs for next middleware/event which will be passed as
90			**kwargs [dict(Any)]
91			)
92
93			Two parameters are passed to the middleware. The first parameter is
94			the current socket connection with and the second one is the payload
95			parameter which is of type :class:`~.core.dispatch.GatewayDispatch`.
96			This contains the response from the discord API.
97
98			:Implementation example:
99
100			.. code-block:: pycon
101
102			>>> @middleware("ready", override=True)
103			>>> async def custom_ready(_, payload: GatewayDispatch):
104			>>> return "on_ready", [User.from_dict(payload.data.get("user"))]
105
106			>>> @Client.event
107			>>> async def on_ready(bot: User):
108			>>> print(f"Signed in as {bot}")
109
110
111			:param call:
112			The call where the method should be registered.
113
114			Keyword Arguments:
115
116			:param override:
117			Setting this to True will allow you to override existing
118			middleware. Usage of this is discouraged, but can help you out
119			of some situations.
120			"""
121
122			def decorator(func: Coro):
123			if override:
124			_log.warning(
125			"Middleware overriding has been enabled for `%s`."
126			" This might cause unexpected behavior.", call
127			)
128
129			if not override and callable(_events.get(call)):
130			raise RuntimeError(
131			f"Middleware event with call `{call}` has "
132			"already been registered"
133			)
134
135			async def wrapper(cls, payload: GatewayDispatch):
136			_log.debug("`%s` middleware has been invoked", call)
137
138			return await (
139			func(cls, payload)
140			if should_pass_cls(func)
141			else await func(payload)
142			)
143
144			_events[call] = wrapper
145			return wrapper
146
147			return decorator
148
149
150			class Client(Dispatcher):
			0 ignored issues – show introduced 2021-09-01 19:26 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
151			def __init__(self, token: str):
152			"""
153			The client is the main instance which is between the programmer
154			and the discord API.
155
156			This client represents your bot.
157
158			:param token:
159			The secret bot token which can be found in
160			`<https://discord.com/developers/applications/\<bot_id\>/bot>`_
161			"""
162			# 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...
163			super().__init__(
164			token,
165			handlers={
166			# Use this event handler for opcode 0.
167			0: self.event_handler
168			}
169			)
170
171			self.bot: Optional[User] = None
172			self.__token = token
173
174			@property
175			def http(self):
176			"""
177			Returns a http client with the current client its
178			authentication credentials.
179
180			:Usage example:
181
182			.. code-block:: pycon
183
184			>>> async with self.http as client:
185			>>> await client.post(
186			... '<endpoint>',
187			... {
188			... "foo": "bar",
189			... "bar": "baz",
190			... "baz": "foo"
191			... }
192			... )
193
194			"""
195			return HTTPClient(self.__token)
196
197			@staticmethod
198			def event(coroutine: Coro):
199			"""
200			Register a Discord gateway event listener. This event will get
201			called when the client receives a new event update from Discord
202			which matches the event name.
203
204			The event name gets pulled from your method name, and this must
205			start with ``on_``. This forces you to write clean and consistent
206			code.
207
208			This decorator can be used in and out of a class, and all
209			event methods must be coroutines. (async)
210
211			:Example usage:
212
213			.. code-block:: pycon
214
215			>>> # Function based
216			>>> from pincer import Client
217			>>>
218			>>> client = Client("token")
219			>>>
220			>>> @client.event
221			>>> async def on_ready():
222			... print(f"Signed in as {client.bot}")
223			>>>
224			>>> if __name__ == "__main__":
225			... client.run()
226
227			.. code-block :: pycon
228
229			>>> # Class based
230			>>> from pincer import Client
231			>>>
232			>>> class BotClient(Client):
233			... @Client.event
234			... async def on_ready(self):
235			... print(f"Signed in as {self.bot}")
236			>>>
237			>>> if __name__ == "__main__":
238			... BotClient("token").run()
239
240
241			:param coroutine: # TODO: add info
242
243			:raises TypeError:
244			If the method is not a coroutine.
245
246			:raises InvalidEventName:
247			If the event name does not start with ``on_``, has already
248			been registered or is not a valid event name.
249			"""
250
251			if not iscoroutinefunction(coroutine):
252			raise TypeError(
253			"Any event which is registered must be a coroutine function"
254			)
255
256			name: str = coroutine.__name__.lower()
257
258			if not name.startswith("on_"):
259			raise InvalidEventName(
260			f"The event `{name}` its name must start with `on_`"
261			)
262
263			if _events.get(name) is not None:
264			raise InvalidEventName(
265			f"The event `{name}` has already been registered or is not "
266			f"a valid event name."
267			)
268
269			_events[name] = coroutine
270			return coroutine
271
272			async def handle_middleware(
273			self,
274			payload: GatewayDispatch,
275			key: str,
276			*args,
277			**kwargs
278			) -> Tuple[Optional[Coro], List[Any], Dict[str, Any]]:
279			"""
280			Handles all middleware recursively. Stops when it has found an
281			event name which starts with ``on_``.
282
283			:param payload:
284			The original payload for the event.
285
286			:param key:
287			The index of the middleware in ``_events``.
288
289			:param \*args:
290			The arguments which will be passed to the middleware.
291
292			:param \\kwargs:
293			The named arguments which will be passed to the middleware.
294
295			:return:
296			A tuple where the first element is the final executor
297			(so the event) its index in ``_events``. The second and third
298			element are the ``args`` and ``*kwargs`` for the event.
299			"""
300			ware: MiddlewareType = _events.get(key)
301			next_call, arguments, params = ware, [], {}
302
303			if iscoroutinefunction(ware):
304			extractable = await ware(self, payload, args, *kwargs)
305
306			if not isinstance(extractable, tuple):
307			raise RuntimeError(
308			f"Return type from `{key}` middleware must be tuple. "
309			)
310
311			next_call = get_index(extractable, 0, "")
312			arguments = get_index(extractable, 1, [])
313			params = get_index(extractable, 2, {})
314
315			if next_call is None:
316			raise RuntimeError(f"Middleware `{key}` has not been registered.")
317
318			return (
319			(next_call, arguments, params)
320			if next_call.startswith("on_")
321			else await self.handle_middleware(
322			payload, next_call, arguments, *params
323			)
324			)
325
326			async def event_handler(self, _, payload: GatewayDispatch):
327			"""
328			Handles all payload events with opcode 0.
329
330			:param _:
331			Socket param, but this isn't required for this handler. So
332			its just a filler parameter, doesn't matter what is passed.
333
334			:param payload:
335			The payload sent from the Discord gateway, this contains the
336			required data for the client to know what event it is and
337			what specifically happened.
338			"""
339			event_name = payload.event_name.lower()
340
341			key, args, kwargs = await self.handle_middleware(payload, event_name)
342
343			call = _events.get(key)
344
345			if iscoroutinefunction(call):
346			if should_pass_cls(call):
347			await call(self, args, *kwargs)
348			else:
349			await call(args, *kwargs)
350
351			@middleware("ready")
352			async def on_ready_middleware(self, payload: GatewayDispatch):
353			"""
354			Middleware for ``on_ready`` event.
355
356			:param payload:
357			The data received from the ready event.
358			"""
359			self.bot = User.from_dict(payload.data.get("user"))
360			return "on_ready",
361
362			@middleware("interaction_create")
363			async def on_interaction_middleware(self, payload: GatewayDispatch):
364			"""
365			Middleware for ``on_interaction``, which handles command
366			execution.
367
368			:param payload:
369			The data received from the interaction event.
370			"""
371			interaction: Interaction = Interaction.from_dict(payload.data)
372			command = ChatCommandHandler.register.get(interaction.data.name)
373
374			if command:
375			kwargs = {opt.name: opt.value for opt in interaction.data.options}
376
377			if should_pass_cls(command.call):
378			kwargs["self"] = self
379
380			res = await command.call(**kwargs)
381
382			if res:
383			async with self.http as http:
384			await http.post(f"interactions/{interaction.id}/{interaction.token}/callback", {
385			"type": 4,
386			"data": {
387			"content": str(res)
388			}
389			})
390
391			return "on_interaction_create", [interaction]
392
393
394			Bot = Client
395

Pincer-org / Pincer

Push — main ( f6a351...3d8f6e )

pincer.client.Client.event() A

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like