pincer.client

Complexity

Total Complexity

19

Size/Duplication

Total Lines	238
Duplicated Lines	0 %

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	Client.event_handler()	0	15	3
A	Client.event()	0	24	4
B	Client.handle_middleware()	0	42	5
A	Client.on_ready_middleware()	0	5	1
A	Client.__init__()	0	20	1

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

Missing module docstring

# 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__

This seems to re-define the built-in __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]]]]

Class name "middleware_type" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern)

_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.

    :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}`."

Use lazy % formatting in logging functions

                         f"This might cause unexpected behaviour.")


        if not override and iscoroutinefunction(_events.get(call)):

The variable _events does not seem to be defined.

The variable call does not seem to be defined.

            raise RuntimeError(f"Middleware event with call `{call}` has "

                               f"already been registered or is no coroutine.")


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

Missing class docstring

    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

TODO and FIXME comments should generally be avoided.

        super().__init__(

            token,

            handlers={

                # Use this event handler for opcode 0.

                0: self.event_handler

            }

        )


        # TODO: close the client after use

TODO and FIXME comments should generally be avoided.

        self.http = HTTPClient(token, version=GatewayConfig.version)

        self.bot: Optional[User] = None


    @staticmethod

    def event(coroutine: Coro):

Missing function or method docstring

        # TODO: Write docs

TODO and FIXME comments should generally be avoided.

        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:

The variable _events does not seem to be defined.

            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]]:

Value 'tuple' is unsubscriptable

        """

        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.

        """

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

                kwargs["self"] = self


            await call(*args, **kwargs)


    @middleware("ready")

    async def on_ready_middleware(self, payload: GatewayDispatch):

        """Middleware for `on_ready` event. """

        self.bot = User.from_dict(payload.data.get("user"))

        return "on_ready",

Disallow trailing comma tuple

Bot = Client