pincer.utils.event_mgr.EventMgr.__init__() - Code Metrics - Inspection of "✨ Added Client.wait_for() and Client.loop_for()" - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#176)

unknown

created 2021-11-06 00:54 UTC

pincer.utils.event_mgr.EventMgr.init() A

↳ Parent: pincer.utils.event_mgr

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	2
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 asyncio import Event, wait_for as _wait_for, get_running_loop
from typing import Any, Callable, Union

from .types import CheckFunction


class _DiscordEvent(Event):
    """
    Attributes
    ----------
    return_value : Optional[str]
        Used to store the arguments from ``can_be_set`` so they can be
        returned later.
    """

    def __init__(
        self,

        event_name: str,

        check: CheckFunction

    ):
        """
        Parameters
        ----------
        event_name : str
            The name of the event.
        check : Optional[Callable[[Any], bool]]
            ``can_be_set`` only returns true if this function returns true.
            Will be ignored if set to None.
        """
        self.event_name = event_name
        self.check = check
        self.return_value = None
        super().__init__()

    def can_be_set(self, event_name: str, *args) -> bool:
        """
        Parameters
        ----------
        event_name : str
            The name of the event.
        *args : Any
            Arguments to evaluate check with.

        Returns
        -------
        bool
            Whether the event can be set
        """
        if self.event_name != event_name:
            return False

        self.return_value = args

        if self.check:
            return self.check(*args)

        return True


class EventMgr:
    """
    Attributes
    ----------
    event_list : List[_DiscordEvent]
        The List of events that need to be processed.
    """

    def __init__(self):
        self.event_list = []

    def add_event(self, event_name: str, check: CheckFunction):
        """
        Parameters
        ----------
        event_name : str
            The type of event to listen for. Uses the same naming scheme as
            @Client.event.
        check : Optional[Callable[[Any], bool]]
            Expression to evaluate when checking if an event is valid. Will
            return be set if this event is true. Will be ignored if set to
            None.

        Returns
        -------
        _DiscordEvent
            Event that was added to the stack.
        """
        event = _DiscordEvent(
            event_name=event_name,
            check=check
        )
        self.event_list.append(event)
        return event

    def process_events(self, event_name, *args):
        """
        Parameters
        ----------
        event_name : str
            The name of the event to be processed.
        *args : Any
            The arguments returned from the middleware for this event.
        """
        for event in self.event_list:
            if event.can_be_set(event_name, *args):
                event.set()

    async def wait_for(
        self,

        event_name: str,

        check: CheckFunction,

        timeout: Union[float, None]

    ) -> Any:
        """
        Parameters
        ----------
        event_name : str
            The type of event. It should start with `on_`. This is the same
            name that is used for @Client.event.
        check : Union[Callable[[Any], bool], None]
            This function only returns a value if this return true.
        timeout: Union[float, None]
            Amount of seconds before timeout. Use None for no timeout.

        Returns
        ------
        Any
            What the Discord API returns for this event.
        """
        event = self.add_event(event_name, check)
        try:
            await _wait_for(event.wait(), timeout=timeout)
        except TimeoutError:
            raise TimeoutError(
                "wait_for() timed out while waiting for an event."
            )
        self.event_list.remove(event)
        return event.return_value

    async def loop_for(
        self,

        event_name: str,

        check: Union[Callable[[Any], bool], None],

        iteration_timeout: Union[float, None],

        loop_timeout: Union[float, None],

    ) -> Any:
        """
        Parameters
        ----------
        event_name : str
            The type of event. It should start with `on_`. This is the same
            name that is used for @Client.event.
        check : Callable[[Any], bool]
            This function only returns a value if this return true.
        iteration_timeout: Union[float, None]
            Amount of seconds before timeout. Timeouts are for each loop.
        loop_timeout: Union[float, None]
            Amount of seconds before the entire loop times out. The generator
            will only raise a timeout error while it is waiting for an event.

        Yields
        ------
        Any
            What the Discord API returns for this event.
        """

        if not loop_timeout:
            while True:
                yield await self.wait_for(event_name, check, iteration_timeout)

        loop = get_running_loop()

        while True:
            start_time = loop.time()

            try:
                yield await _wait_for(
                    self.wait_for(
                        event_name,
                        check,
                        iteration_timeout
                    ),
                    timeout=loop_timeout
                )

            except TimeoutError:
                raise TimeoutError(
                    "loop_for() timed out while waiting for an event"
                )

            loop_timeout -= loop.time() - start_time

            # loop_timeout can be below 0 if the user's code in the for loop
            # takes longer than the time left in loop_timeout
            if loop_timeout <= 0:
                break


1			# Copyright Pincer 2021-Present
			0 ignored issues – show introduced 2021-09-25 13:19 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 asyncio import Event, wait_for as _wait_for, get_running_loop
5			from typing import Any, Callable, Union
6
7			from .types import CheckFunction
8
9
10			class _DiscordEvent(Event):
11			"""
12			Attributes
13			----------
14			return_value : Optional[str]
15			Used to store the arguments from ``can_be_set`` so they can be
16			returned later.
17			"""
18
19			def __init__(
20			self,
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
21			event_name: str,
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
22			check: CheckFunction
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
23			):
24			"""
25			Parameters
26			----------
27			event_name : str
28			The name of the event.
29			check : Optional[Callable[[Any], bool]]
30			``can_be_set`` only returns true if this function returns true.
31			Will be ignored if set to None.
32			"""
33			self.event_name = event_name
34			self.check = check
35			self.return_value = None
36			super().__init__()
37
38			def can_be_set(self, event_name: str, *args) -> bool:
39			"""
40			Parameters
41			----------
42			event_name : str
43			The name of the event.
44			*args : Any
45			Arguments to evaluate check with.
46
47			Returns
48			-------
49			bool
50			Whether the event can be set
51			"""
52			if self.event_name != event_name:
53			return False
54
55			self.return_value = args
56
57			if self.check:
58			return self.check(*args)
59
60			return True
61
62
63			class EventMgr:
64			"""
65			Attributes
66			----------
67			event_list : List[_DiscordEvent]
68			The List of events that need to be processed.
69			"""
70
71			def __init__(self):
72			self.event_list = []
73
74			def add_event(self, event_name: str, check: CheckFunction):
75			"""
76			Parameters
77			----------
78			event_name : str
79			The type of event to listen for. Uses the same naming scheme as
80			@Client.event.
81			check : Optional[Callable[[Any], bool]]
82			Expression to evaluate when checking if an event is valid. Will
83			return be set if this event is true. Will be ignored if set to
84			None.
85
86			Returns
87			-------
88			_DiscordEvent
89			Event that was added to the stack.
90			"""
91			event = _DiscordEvent(
92			event_name=event_name,
93			check=check
94			)
95			self.event_list.append(event)
96			return event
97
98			def process_events(self, event_name, *args):
99			"""
100			Parameters
101			----------
102			event_name : str
103			The name of the event to be processed.
104			*args : Any
105			The arguments returned from the middleware for this event.
106			"""
107			for event in self.event_list:
108			if event.can_be_set(event_name, *args):
109			event.set()
110
111			async def wait_for(
112			self,
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
113			event_name: str,
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
114			check: CheckFunction,
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
115			timeout: Union[float, None]
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
116			) -> Any:
117			"""
118			Parameters
119			----------
120			event_name : str
121			The type of event. It should start with `on_`. This is the same
122			name that is used for @Client.event.
123			check : Union[Callable[[Any], bool], None]
124			This function only returns a value if this return true.
125			timeout: Union[float, None]
126			Amount of seconds before timeout. Use None for no timeout.
127
128			Returns
129			------
130			Any
131			What the Discord API returns for this event.
132			"""
133			event = self.add_event(event_name, check)
134			try:
135			await _wait_for(event.wait(), timeout=timeout)
136			except TimeoutError:
137			raise TimeoutError(
138			"wait_for() timed out while waiting for an event."
139			)
140			self.event_list.remove(event)
141			return event.return_value
142
143			async def loop_for(
144			self,
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
145			event_name: str,
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
146			check: Union[Callable[[Any], bool], None],
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
147			iteration_timeout: Union[float, None],
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
148			loop_timeout: Union[float, None],
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
149			) -> Any:
150			"""
151			Parameters
152			----------
153			event_name : str
154			The type of event. It should start with `on_`. This is the same
155			name that is used for @Client.event.
156			check : Callable[[Any], bool]
157			This function only returns a value if this return true.
158			iteration_timeout: Union[float, None]
159			Amount of seconds before timeout. Timeouts are for each loop.
160			loop_timeout: Union[float, None]
161			Amount of seconds before the entire loop times out. The generator
162			will only raise a timeout error while it is waiting for an event.
163
164			Yields
165			------
166			Any
167			What the Discord API returns for this event.
168			"""
169
170			if not loop_timeout:
171			while True:
172			yield await self.wait_for(event_name, check, iteration_timeout)
173
174			loop = get_running_loop()
175
176			while True:
177			start_time = loop.time()
178
179			try:
180			yield await _wait_for(
181			self.wait_for(
182			event_name,
183			check,
184			iteration_timeout
185			),
186			timeout=loop_timeout
187			)
188
189			except TimeoutError:
190			raise TimeoutError(
191			"loop_for() timed out while waiting for an event"
192			)
193
194			loop_timeout -= loop.time() - start_time
195
196			# loop_timeout can be below 0 if the user's code in the for loop
197			# takes longer than the time left in loop_timeout
198			if loop_timeout <= 0:
199			break
200

Pincer-org / Pincer

Pull Request — main (#176)

pincer.utils.event_mgr.EventMgr.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

pincer.utils.event_mgr.EventMgr.init() A