pincer.utils.tasks.Task.client_required() - Code Metrics - Inspection of "Upgrade docs" - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#155)

by Oliver

created 2021-10-24 15:39 UTC

pincer.utils.tasks.Task.client_required() A

↳ Parent: pincer.utils.tasks.TaskScheduler.loop()

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	3
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 asyncio
import logging
from datetime import timedelta
from typing import TYPE_CHECKING
from asyncio import TimerHandle, iscoroutinefunction

from .types import Coro
from . import __package__
from .insertion import should_pass_cls

if TYPE_CHECKING:
    from typing import Callable, Set


    from .types import Coro

    from ..exceptions import (
        TaskAlreadyRunning, TaskCancelError, TaskInvalidDelay,
        TaskIsNotCoroutine
    )
    from .insertion import should_pass_cls


_log = logging.getLogger(__package__)


class TaskScheduler:
    """Class that scedules tasts.
    """

    def __init__(self, client):
        self.client = client
        self.tasks: Set[Task] = set()
        self._loop = asyncio.get_event_loop()

    def loop(

        self,

        days=0,

        weeks=0,

        hours=0,

        minutes=0,

        seconds=0,

        milliseconds=0,

        microseconds=0

    ) -> Callable[[Coro], Task]:
        """A decorator to create a task that repeat the given amount of t

        :Example usage:

        .. code-block:: python

            from pincer import Client
            from pincer.utils import TaskScheduler

            client = Client("token")
            task = TaskScheduler(client)

            @task.loop(minutes=3)
            async def my_task(self):
                ...

            my_task.start()
            client.run()

        Parameters
        ----------
        days : :class:`int`
            Days to wait between iterations.
            |default| ``0``
        weeks : :class:`int`
            Days to wait between iterations.
            |default| ``0``
        hours : :class:`int`
            Days to wait between iterations.
            |default| ``0``
        minutes : :class:`int`
            Days to wait between iterations.
            |default| ``0``
        seconds : :class:`int`
            Days to wait between iterations.
            |default| ``0``
        milliseconds : :class:`int`
            Days to wait between iterations.
            |default| ``0``
        microseconds : :class:`int`
            Days to wait between iterations.
            |default| ``0``

        Raises
        ------
        TaskIsNotCoroutine:
            The task is not a coroutine.
        TaskInvalidDelay:
            The delay is 0 or negative.
        """
        def decorator(func: Coro) -> Task:
            if not iscoroutinefunction(func):
                raise TaskIsNotCoroutine(

                    f'Task `{func.__name__}` is not a coroutine, '
                    'which is required for tasks.'
                )

            delay = timedelta(
                days=days,
                weeks=weeks,
                hours=hours,
                minutes=minutes,
                seconds=seconds,
                microseconds=microseconds,
                milliseconds=milliseconds
            ).total_seconds()

            if delay <= 0:
                raise TaskInvalidDelay(

                    f'Task `{func.__name__}` has a delay of {delay} seconds, '
                    'which is invalid. Delay must be greater than zero.'
                )

            return Task(self, func, delay)

        return decorator

    def register(self, task: Task):
        """Register a task.

        Parameters
        ----------
        task : :class:`~pincer.utils.tasks.Task`
            The task to register.
        """
        self.tasks.add(task)
        self.__execute(task)

    def __execute(self, task: Task):
        """Execute a task."""
        coro = task.coro(self.client) if task.client_required else task.coro()
        # Execute the coroutine
        asyncio.ensure_future(coro)

        # Schedule the coroutine's next execution
        task._handle = self._loop.call_later(task.delay, self.__execute, task)
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent

    def close(self):
        """Gracefully stops any running task."""
        for task in self.tasks.copy():
            if task.running:
                task.cancel()


class Task:
    """A Task is a coroutine that is scheduled to repeat every x seconds.
    Use a TaskScheduler in order to create a task.

    Parameters
    ----------
    scheduler: :class:`~pincer.utils.tasks.TaskScheduler`
        The scheduler to use.
    coro: :class:`~pincer.utils.types.Coro`
        The coroutine to register as a task.
    delay: :class:`float`
        Delay between each iteration of the task.
    """

    def __init__(self, scheduler: TaskScheduler, coro: Coro, delay: float):
        self._scheduler = scheduler
        self.coro = coro
        self.delay = delay
        self._handle: TimerHandle = None
        self._client_required = should_pass_cls(coro)

    def __del__(self):
        if self.running:
            self.cancel()
        else:
            # Did the user forgot to call task.start() ?
            _log.warn(

                "Task `%s` was not scheduled. Did you forget to start it ?",
                self.coro.__name__
            )

    @property
    def cancelled(self):
        """:class:`bool`: Check if the task has been cancelled or not.
        """
        return self.running and self._handle.cancelled()

    @property
    def running(self):
        """:class:`bool`: Check if the task is running.
        """
        return self._handle is not None

    def start(self):
        """Register the task in the TaskScheduler and start
        the execution of the task.
        """
        if self.running:
            raise TaskAlreadyRunning(

                f'Task `{self.coro.__name__}` is already running.', self
            )

        self._scheduler.register(self)

    def cancel(self):
        """Cancel the task.
        """
        if not self.running:
            raise TaskCancelError(

                f'Task `{self.coro.__name__}` is not running.', self
            )

        self._handle.cancel()
        if self in self._scheduler.tasks:
            self._scheduler.tasks.remove(self)


1			# Copyright Pincer 2021-Present
			0 ignored issues – show introduced 2021-10-03 07:58 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 asyncio
7			import logging
8			from datetime import timedelta
9			from typing import TYPE_CHECKING
10			from asyncio import TimerHandle, iscoroutinefunction
11
12			from .types import Coro
13			from . import __package__
14			from .insertion import should_pass_cls
15
16			if TYPE_CHECKING:
17			from typing import Callable, Set
			0 ignored issues – show introduced 2021-10-23 19:39 UTC by Report Bug Copy Issue Report Imports from package typing are not grouped Loading history...
18
19			from .types import Coro
			0 ignored issues – show Unused Code introduced 2021-10-23 19:39 UTC by Report Bug Copy Issue Report The import `Coro` was already done on line 12. You should be able to remove this line. Loading history...
20			from ..exceptions import (
21			TaskAlreadyRunning, TaskCancelError, TaskInvalidDelay,
22			TaskIsNotCoroutine
23			)
24			from .insertion import should_pass_cls
			0 ignored issues – show Unused Code introduced 2021-10-23 19:39 UTC by Report Bug Copy Issue Report The import `should_pass_cls` was already done on line 14. You should be able to remove this line. Loading history...
25
26			_log = logging.getLogger(__package__)
27
28
29			class TaskScheduler:
30			"""Class that scedules tasts.
31			"""
32
33			def __init__(self, client):
34			self.client = client
35			self.tasks: Set[Task] = set()
36			self._loop = asyncio.get_event_loop()
37
38			def loop(
			0 ignored issues – show best-practice introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Too many arguments (8/5) Loading history...
39			self,
			0 ignored issues – show Coding Style introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
40			days=0,
			0 ignored issues – show Coding Style introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
41			weeks=0,
			0 ignored issues – show Coding Style introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
42			hours=0,
			0 ignored issues – show Coding Style introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
43			minutes=0,
			0 ignored issues – show Coding Style introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
44			seconds=0,
			0 ignored issues – show Coding Style introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
45			milliseconds=0,
			0 ignored issues – show Coding Style introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
46			microseconds=0
			0 ignored issues – show Coding Style introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
47			) -> Callable[[Coro], Task]:
48			"""A decorator to create a task that repeat the given amount of t
49
50			:Example usage:
51
52			.. code-block:: python
53
54			from pincer import Client
55			from pincer.utils import TaskScheduler
56
57			client = Client("token")
58			task = TaskScheduler(client)
59
60			@task.loop(minutes=3)
61			async def my_task(self):
62			...
63
64			my_task.start()
65			client.run()
66
67			Parameters
68			----------
69			days : :class:`int`
70			Days to wait between iterations.
71			\|default\| ``0``
72			weeks : :class:`int`
73			Days to wait between iterations.
74			\|default\| ``0``
75			hours : :class:`int`
76			Days to wait between iterations.
77			\|default\| ``0``
78			minutes : :class:`int`
79			Days to wait between iterations.
80			\|default\| ``0``
81			seconds : :class:`int`
82			Days to wait between iterations.
83			\|default\| ``0``
84			milliseconds : :class:`int`
85			Days to wait between iterations.
86			\|default\| ``0``
87			microseconds : :class:`int`
88			Days to wait between iterations.
89			\|default\| ``0``
90
91			Raises
92			------
93			TaskIsNotCoroutine:
94			The task is not a coroutine.
95			TaskInvalidDelay:
96			The delay is 0 or negative.
97			"""
98			def decorator(func: Coro) -> Task:
99			if not iscoroutinefunction(func):
100			raise TaskIsNotCoroutine(
			0 ignored issues – show Bug introduced 2021-10-18 18:08 UTC by Report Bug Copy Issue Report The variable `TaskIsNotCoroutine` was used before it was assigned. Loading history... introduced 2021-10-18 18:08 UTC by Report Bug Copy Issue Report The variable `TaskIsNotCoroutine` does not seem to be defined in case `TYPE_CHECKING` on line `16` is `False`. Are you sure this can never be the case? Loading history...
101			f'Task `{func.__name__}` is not a coroutine, '
102			'which is required for tasks.'
103			)
104
105			delay = timedelta(
106			days=days,
107			weeks=weeks,
108			hours=hours,
109			minutes=minutes,
110			seconds=seconds,
111			microseconds=microseconds,
112			milliseconds=milliseconds
113			).total_seconds()
114
115			if delay <= 0:
116			raise TaskInvalidDelay(
			0 ignored issues – show introduced 2021-10-18 18:08 UTC by Report Bug Copy Issue Report The variable `TaskInvalidDelay` does not seem to be defined in case `TYPE_CHECKING` on line `16` is `False`. Are you sure this can never be the case? Loading history... Bug introduced 2021-10-18 18:08 UTC by Report Bug Copy Issue Report The variable `TaskInvalidDelay` was used before it was assigned. Loading history...
117			f'Task `{func.__name__}` has a delay of {delay} seconds, '
118			'which is invalid. Delay must be greater than zero.'
119			)
120
121			return Task(self, func, delay)
122
123			return decorator
124
125			def register(self, task: Task):
126			"""Register a task.
127
128			Parameters
129			----------
130			task : :class:`~pincer.utils.tasks.Task`
131			The task to register.
132			"""
133			self.tasks.add(task)
134			self.__execute(task)
135
136			def __execute(self, task: Task):
137			"""Execute a task."""
138			coro = task.coro(self.client) if task.client_required else task.coro()
139			# Execute the coroutine
140			asyncio.ensure_future(coro)
141
142			# Schedule the coroutine's next execution
143			task._handle = self._loop.call_later(task.delay, self.__execute, task)
			0 ignored issues – show Coding Style Best Practice introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report It seems like `_handle` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
144
145			def close(self):
146			"""Gracefully stops any running task."""
147			for task in self.tasks.copy():
148			if task.running:
149			task.cancel()
150
151
152			class Task:
153			"""A Task is a coroutine that is scheduled to repeat every x seconds.
154			Use a TaskScheduler in order to create a task.
155
156			Parameters
157			----------
158			scheduler: :class:`~pincer.utils.tasks.TaskScheduler`
159			The scheduler to use.
160			coro: :class:`~pincer.utils.types.Coro`
161			The coroutine to register as a task.
162			delay: :class:`float`
163			Delay between each iteration of the task.
164			"""
165
166			def __init__(self, scheduler: TaskScheduler, coro: Coro, delay: float):
167			self._scheduler = scheduler
168			self.coro = coro
169			self.delay = delay
170			self._handle: TimerHandle = None
171			self._client_required = should_pass_cls(coro)
172
173			def __del__(self):
174			if self.running:
175			self.cancel()
176			else:
177			# Did the user forgot to call task.start() ?
178			_log.warn(
			0 ignored issues – show introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Using deprecated method warn() Loading history...
179			"Task `%s` was not scheduled. Did you forget to start it ?",
180			self.coro.__name__
181			)
182
183			@property
184			def cancelled(self):
185			""":class:`bool`: Check if the task has been cancelled or not.
186			"""
187			return self.running and self._handle.cancelled()
188
189			@property
190			def running(self):
191			""":class:`bool`: Check if the task is running.
192			"""
193			return self._handle is not None
194
195			def start(self):
196			"""Register the task in the TaskScheduler and start
197			the execution of the task.
198			"""
199			if self.running:
200			raise TaskAlreadyRunning(
			0 ignored issues – show introduced 2021-10-18 18:08 UTC by Report Bug Copy Issue Report The variable `TaskAlreadyRunning` does not seem to be defined in case `TYPE_CHECKING` on line `16` is `False`. Are you sure this can never be the case? Loading history... Bug introduced 2021-10-18 18:08 UTC by Report Bug Copy Issue Report The variable `TaskAlreadyRunning` was used before it was assigned. Loading history...
201			f'Task `{self.coro.__name__}` is already running.', self
202			)
203
204			self._scheduler.register(self)
205
206			def cancel(self):
207			"""Cancel the task.
208			"""
209			if not self.running:
210			raise TaskCancelError(
			0 ignored issues – show introduced 2021-10-18 18:08 UTC by Report Bug Copy Issue Report The variable `TaskCancelError` does not seem to be defined in case `TYPE_CHECKING` on line `16` is `False`. Are you sure this can never be the case? Loading history... Bug introduced 2021-10-18 18:08 UTC by Report Bug Copy Issue Report The variable `TaskCancelError` was used before it was assigned. Loading history...
211			f'Task `{self.coro.__name__}` is not running.', self
212			)
213
214			self._handle.cancel()
215			if self in self._scheduler.tasks:
216			self._scheduler.tasks.remove(self)
217

Pincer-org / Pincer

Pull Request — main (#155)

pincer.utils.tasks.Task.client_required() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like