pincer.client.Client._http() - Code Metrics - Inspection of ":memo: Update documentation" - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 1ee03a...dea977 )

unknown

created 2021-09-03 10:23 UTC

pincer.client.Client._http() A

↳ Parent: pincer.client

Complexity

Conditions

Size

Total Lines	19
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- 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, Any, Union, Dict, Tuple, List

from pincer import __package__

from pincer._config import GatewayConfig, events

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
from pincer.utils.extraction import get_index
from pincer.utils.insertion import should_pass_cls
from pincer.utils.types import Coro

_log = logging.getLogger(__package__)

middleware_type = 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:
    ```
        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)]
        )
    ```

    One parameter is passed to the middleware. This parameter is the
    payload parameter which is of type GatewayDispatch. This contains
    the response from the discord API.

    Implementation example:
    ```py
    >>> @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(f"Middleware overriding has been enabled for `{call}`."

                         " This might cause unexpected behaviour.")

        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:
        ```py
        >>> 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:
        ```py
        >>> # 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()
        ```
        ```py
        >>> # 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()
        ```

        :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: middleware_type = _events.get(key)
        next_call, arguments, params = ware, list(), dict()

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

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

            next_call = get_index(extractable, 0, "")
            arguments = get_index(extractable, 1, list())
            params = get_index(extractable, 2, dict())

        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",



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, Any, Union, Dict, Tuple, List
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, events
			0 ignored issues – show Unused Code introduced 2021-09-03 10:24 UTC by Report Bug Copy Issue Report Unused GatewayConfig imported from pincer._config Loading history...
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			from pincer.utils.extraction import get_index
36			from pincer.utils.insertion import should_pass_cls
37			from pincer.utils.types import Coro
38
39			_log = logging.getLogger(__package__)
40
41			middleware_type = Optional[Union[Coro, Tuple[str, List[Any], Dict[str, Any]]]]
			1 ignored issue – show Coding Style Naming introduced 2021-09-01 19:26 UTC by Report Bug Copy Issue Report Class name "middleware_type" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
42
43			_events: Dict[str, Optional[Union[str, Coro]]] = {}
44
45			for event in events:
46			event_final_executor = f"on_{event}"
47
48			# Event middleware for the library. Function argument is a payload
49			# (GatewayDispatch). The function must return a string which
50			# contains the main event key. As second value a list with arguments,
51			# and thee third value value must be a dictionary. The last two are
52			# passed on as args and *kwargs.
53			#
54			# NOTE: These return values must be passed as a tuple!
55			_events[event] = event_final_executor
56
57			# The registered event by the client. Do not manually overwrite.
58			_events[event_final_executor] = None
59
60
61			def middleware(call: str, *, override: bool = False):
62			"""
63			Middleware are methods which can be registered with this decorator.
64			These methods are invoked before any `on_` event. As the `on_` event
65			is the final call.
66
67			A default call exists for all events, but some might already be in
68			use by the library. If you know what you are doing, you can override
69			these default middleware methods by passing the override parameter.
70
71			The method to which this decorator is registered must be a coroutine,
72			and it must return a tuple with the following format:
73			```
74			tuple(
75			key for next middleware or final event [str],
76			args for next middleware/event which will be passed as *args
77			[list(Any)],
78			kwargs for next middleware/event which will be passed as
79			**kwargs [dict(Any)]
80			)
81			```
82
83			One parameter is passed to the middleware. This parameter is the
84			payload parameter which is of type GatewayDispatch. This contains
85			the response from the discord API.
86
87			Implementation example:
88			```py
89			>>> @middleware("ready", override=True)
90			>>> async def custom_ready(payload: GatewayDispatch):
91			>>> return "on_ready", [User.from_dict(payload.data.get("user"))]
92
93			>>> @Client.event
94			>>> async def on_ready(bot: User):
95			>>> print(f"Signed in as {bot}")
96			```
97
98			:param call:
99			The call where the method should be registered.
100
101			Keyword Arguments:
102
103			:param override:
104			Setting this to True will allow you to override existing
105			middleware. Usage of this is discouraged, but can help you out
106			of some situations.
107			"""
108			def decorator(func: Coro):
109			if override:
110			_log.warning(f"Middleware overriding has been enabled for `{call}`."
			0 ignored issues – show introduced 2021-09-02 11:34 UTC by Report Bug Copy Issue Report Use lazy % formatting in logging functions Loading history...
111			" This might cause unexpected behaviour.")
112
113			if not override and callable(_events.get(call)):
			2 ignored issues – show Comprehensibility Best Practice introduced 2021-08-31 21:18 UTC by Report Bug Copy Issue Report The variable `_events` does not seem to be defined. Loading history... Comprehensibility Best Practice introduced 2021-09-02 11:34 UTC by Report Bug Copy Issue Report The variable `call` does not seem to be defined. Loading history...
114			raise RuntimeError(f"Middleware event with call `{call}` has "
115			"already been registered")
116
117			async def wrapper(cls, payload: GatewayDispatch):
118			_log.debug("`%s` middleware has been invoked", call)
119			return await func(cls, payload) \
120			if should_pass_cls(func) \
121			else await func(payload)
122
123			_events[call] = wrapper
124			return wrapper
125
126			return decorator
127
128
129			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...
130			def __init__(self, token: str):
131			"""
132			The client is the main instance which is between the programmer and the
133			discord API. This client represents your bot.
134
135			:param token:
136			The secret bot token which can be found in
137			https://discord.com/developers/applications/<bot_id>/bot
138			"""
139			# 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...
140			super().__init__(
141			token,
142			handlers={
143			# Use this event handler for opcode 0.
144			0: self.event_handler
145			}
146			)
147
148			self.bot: Optional[User] = None
149			self.__token = token
150
151			@property
152			def _http(self):
153			"""
154			Returns a http client with the current client its
155			authentication credentials.
156
157			Usage example:
158			```py
159			>>> async with self._http as client:
160			>>> await client.post(
161			>>> '<endpoint>',
162			>>> {
163			>>> "foo": "bar",
164			>>> "bar": "baz",
165			>>> "baz": "foo"
166			>>> })
167			```
168			"""
169			return HTTPClient(self.__token)
170
171			@staticmethod
172			def event(coroutine: Coro):
173			"""
174			Register a Discord gateway event listener. This event will get
175			called when the client receives a new event update from Discord
176			which matches the event name.
177
178			The event name gets pulled from your method name, and this must
179			start with `on_`. This forces you to write clean and consistent
180			code.
181
182			This decorator can be used in and out of a class, and all
183			event methods must be coroutines. (async)
184
185			Example usage:
186			```py
187			>>> # Function based
188			>>> from pincer import Client
189			>>>
190			>>> client = Client("token")
191			>>>
192			>>> @client.event
193			>>> async def on_ready():
194			>>> print(f"Signed in as {client.bot}")
195			>>>
196			>>> if __name__ == "__main__":
197			>>> client.run()
198			```
199			```py
200			>>> # Class based
201			>>> from pincer import Client
202			>>>
203			>>> class BotClient(Client):
204			>>> @Client.event
205			>>> async def on_ready(self):
206			>>> print(f"Signed in as {self.bot}")
207			>>>
208			>>> if __name__ == "__main__":
209			>>> BotClient("token").run()
210			```
211
212			:raises TypeError:
213			If the method is not a coroutine.
214
215			:raises InvalidEventName:
216			If the event name does not start with 'on_', has already
217			been registered or is not a valid event name.
218			"""
219
220			if not iscoroutinefunction(coroutine):
221			raise TypeError(
222			"Any event which is registered must be a coroutine function"
223			)
224
225			name: str = coroutine.__name__.lower()
226
227			if not name.startswith("on_"):
228			raise InvalidEventName(
229			f"The event `{name}` its name must start with `on_`"
230			)
231
232			if _events.get(name) is not None:
			1 ignored issue – show Comprehensibility Best Practice introduced 2021-09-02 11:34 UTC by Report Bug Copy Issue Report The variable `_events` does not seem to be defined. Loading history...
233			raise InvalidEventName(
234			f"The event `{name}` has already been registered or is not "
235			f"a valid event name."
236			)
237
238			_events[name] = coroutine
239			return coroutine
240
241			async def handle_middleware(
242			self,
243			payload: GatewayDispatch,
244			key: str,
245			*args,
246			**kwargs
247			) -> tuple[Optional[Coro], List[Any], Dict[str, Any]]:
			0 ignored issues – show introduced 2021-09-02 11:34 UTC by Report Bug Copy Issue Report Value 'tuple' is unsubscriptable Loading history...
248			"""
249			Handles all middleware recursively. Stops when it has found an
250			event name which starts with "on_".
251
252			:param payload:
253			The original payload for the event.
254
255			:param key:
256			The index of the middleware in `_events`.
257
258			:param *args:
259			The arguments which will be passed to the middleware.
260
261			:param **kwargs:
262			The named arguments which will be passed to the middleware.
263
264			:return:
265			A tuple where the first element is the final executor
266			(so the event) its index in `_events`. The second and third
267			element are the `args` and `*kwargs` for the event.
268			"""
269			ware: middleware_type = _events.get(key)
270			next_call, arguments, params = ware, list(), dict()
271
272			if iscoroutinefunction(ware):
273			extractable = await ware(self, payload, args, *kwargs)
274
275			if not isinstance(extractable, tuple):
276			raise RuntimeError(f"Return type from `{key}` middleware must "
277			f"be tuple. ")
278
279			next_call = get_index(extractable, 0, "")
280			arguments = get_index(extractable, 1, list())
281			params = get_index(extractable, 2, dict())
282
283			if next_call is None:
284			raise RuntimeError(f"Middleware `{key}` has not been registered.")
285
286			return (next_call, arguments, params) \
287			if next_call.startswith("on_") \
288			else await self.handle_middleware(payload, next_call,
289			arguments, *params)
290
291			async def event_handler(self, _, payload: GatewayDispatch):
292			"""
293			Handles all payload events with opcode 0.
294
295			:param _:
296			Socket param, but this isn't required for this handler. So
297			its just a filler parameter, doesn't matter what is passed.
298
299			:param payload:
300			The payload sent from the Discord gateway, this contains the
301			required data for the client to know what event it is and
302			what specifically happened.
303			"""
304			event_name = payload.event_name.lower()
305
306			key, args, kwargs = await self.handle_middleware(payload, event_name)
307
308			call = _events.get(key)
309
310			if iscoroutinefunction(call):
311			if should_pass_cls(call):
312			await call(self, args, *kwargs)
313			else:
314			await call(args, *kwargs)
315
316			@middleware("ready")
317			async def on_ready_middleware(self, payload: GatewayDispatch):
318			"""
319			Middleware for `on_ready` event.
320
321			:param payload: The data received from the ready event.
322			"""
323			self.bot = User.from_dict(payload.data.get("user"))
324			return "on_ready",
			1 ignored issue – show Unused Code introduced 2021-09-01 19:26 UTC by Report Bug Copy Issue Report Disallow trailing comma tuple Loading history...
325
326
327			Bot = Client
328

Pincer-org / Pincer

Push — main ( 1ee03a...dea977 )

pincer.client.Client._http() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like