pincer.client.Client.chat_commands() - Code Metrics - Inspection of "Merge pull request #75 from Sly-Little-Fox/main" - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — main ( 7f5a9f...48eb55 )

unknown

created 2021-09-25 15:23 UTC

pincer.client.Client.chat_commands() A

↳ Parent: pincer.client

Complexity

Conditions

Size

Total Lines	7
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# Copyright Pincer 2021-Present

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

from __future__ import annotations

import logging
from asyncio import iscoroutinefunction, run
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 .middleware import middleware
from .objects import User, Intents, Guild
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 the 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 event_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

        >>> @event_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)

            # print(func, should_pass_cls(func))

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

        _events[call] = wrapper
        return wrapper

    return decorator


for event, middleware in middleware.items():
    event_middleware(event)(middleware)


class Client(Dispatcher):

    def __init__(
            self,
            token: str, *,
            received: str = None,
            intents: Intents = None
    ):
        """
        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>`_

        :param received:
            The default message which will be sent when no response is
            given.

        :param intents:
            The discord intents for your client.
        """
        super().__init__(
            token,
            handlers={
                # Use this event handler for opcode 0.
                -1: self.payload_event_handler,
                0: self.event_handler
            },
            intents=intents or Intents.NONE,
        )

        self.bot: Optional[User] = None
        self.received_message = received or "Command arrived successfully!"
        self.http = HTTPClient(token)

    @property
    def chat_commands(self):
        """
        Get a list of chat command calls which have been registered in
        the ChatCommandHandler.
        """
        return [cmd.app.name for cmd in ChatCommandHandler.register.values()]

    @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 named `{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

    def run(self):
        """Start the event listener"""
        self.start_loop()
        run(self.http.close())

    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 process_event(self, event_name: str, payload: GatewayDispatch):
        """
        Processes and invokes an event and its middleware.

        :param event_name:
            The name of the event, this is also the filename in the
            middleware directory.

        :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.
        """
        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)

    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.
        """
        await self.process_event(payload.event_name.lower(), payload)

    async def payload_event_handler(self, _, payload: GatewayDispatch):
        """
        Special event which activates on_payload event!

        :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.
        """
        await self.process_event("payload", payload)

    async def get_guild(self, guild_id: int) -> Guild:
        """
        Fetch a guild object by the guild identifier.

        :param guild_id:
            The id of the guild which should be fetched from the Discord
            gateway.

        :returns:
            A Guild object.
        """
        return await Guild.from_id(self, guild_id)

    async def get_user(self, _id: int) -> User:
        """
        Fetch a User from its identifier

        :param _id:
            The id of the user which should be fetched from the Discord
            gateway.

        :returns:
            A User object.
        """
        return await User.from_id(self, _id)


Bot = Client


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

Pincer-org / Pincer

Push — main ( 7f5a9f...48eb55 )

pincer.client.Client.chat_commands() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like