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

Completed

Push — main ( e04df9...e9b147 )

by Yohann

created 2021-11-08 15:05 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 asyncio import TimerHandle, iscoroutinefunction
from datetime import timedelta
from typing import TYPE_CHECKING, Optional

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

if TYPE_CHECKING:
    from typing import Callable, Set
    from .types import Coro


_log = logging.getLogger(__package__)

class TaskScheduler:

    def __init__(self, client):
        """
        Used to create tasks
        """
        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."""
        if task._client_required:
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
            coro = task.coro(self.client)
        else:
            coro = 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: Optional[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.warning(
                "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 asyncio import TimerHandle, iscoroutinefunction
9			from datetime import timedelta
10			from typing import TYPE_CHECKING, Optional
11
12			from . import __package__
13			from .insertion import should_pass_cls
14			from ..exceptions import (
15			TaskAlreadyRunning, TaskCancelError, TaskInvalidDelay,
16			TaskIsNotCoroutine
17			)
18
19			if TYPE_CHECKING:
20			from typing import Callable, Set
21			from .types import Coro
22
23
24			_log = logging.getLogger(__package__)
25
26			class TaskScheduler:
			0 ignored issues – show introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
27			def __init__(self, client):
28			"""
29			Used to create tasks
30			"""
31			self.client = client
32			self.tasks: Set[Task] = set()
33			self._loop = asyncio.get_event_loop()
34
35			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...
36			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...
37			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...
38			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...
39			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...
40			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...
41			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...
42			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...
43			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...
44			) -> Callable[[Coro], Task]:
45			"""A decorator to create a task that repeat the given amount of t
46			:Example usage:
47
48			.. code-block:: python
49			from pincer import Client
50			from pincer.utils import TaskScheduler
51
52			client = Client("token")
53			task = TaskScheduler(client)
54
55			@task.loop(minutes=3)
56			async def my_task(self):
57			...
58
59			my_task.start()
60			client.run()
61			Parameters
62			----------
63			days : :class:`int`
64			Days to wait between iterations.
65			\|default\| ``0``
66			weeks : :class:`int`
67			Days to wait between iterations.
68			\|default\| ``0``
69			hours : :class:`int`
70			Days to wait between iterations.
71			\|default\| ``0``
72			minutes : :class:`int`
73			Days to wait between iterations.
74			\|default\| ``0``
75			seconds : :class:`int`
76			Days to wait between iterations.
77			\|default\| ``0``
78			milliseconds : :class:`int`
79			Days to wait between iterations.
80			\|default\| ``0``
81			microseconds : :class:`int`
82			Days to wait between iterations.
83			\|default\| ``0``
84			Raises
85			------
86			TaskIsNotCoroutine:
87			The task is not a coroutine.
88			TaskInvalidDelay:
89			The delay is 0 or negative.
90			"""
91			def decorator(func: Coro) -> Task:
92			if not iscoroutinefunction(func):
93			raise TaskIsNotCoroutine(
94			f'Task `{func.__name__}` is not a coroutine, '
95			'which is required for tasks.'
96			)
97
98			delay = timedelta(
99			days=days,
100			weeks=weeks,
101			hours=hours,
102			minutes=minutes,
103			seconds=seconds,
104			microseconds=microseconds,
105			milliseconds=milliseconds
106			).total_seconds()
107
108			if delay <= 0:
109			raise TaskInvalidDelay(
110			f'Task `{func.__name__}` has a delay of {delay} seconds, '
111			'which is invalid. Delay must be greater than zero.'
112			)
113
114			return Task(self, func, delay)
115
116			return decorator
117
118			def register(self, task: Task):
119			"""Register a task.
120			Parameters
121			----------
122			task : :class:`~pincer.utils.tasks.Task`
123			The task to register.
124			"""
125			self.tasks.add(task)
126			self.__execute(task)
127
128			def __execute(self, task: Task):
129			"""Execute a task."""
130			if task._client_required:
			0 ignored issues – show Coding Style Best Practice introduced 2021-10-02 17:17 UTC by Report Bug Copy Issue Report It seems like `_client_required` 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...
131			coro = task.coro(self.client)
132			else:
133			coro = task.coro()
134
135			# Execute the coroutine
136			asyncio.ensure_future(coro)
137
138			# Schedule the coroutine's next execution
139			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...
140
141			def close(self):
142			"""Gracefully stops any running task."""
143			for task in self.tasks.copy():
144			if task.running:
145			task.cancel()
146
147			class Task:
148			"""A Task is a coroutine that is scheduled to repeat every x seconds.
149			Use a TaskScheduler in order to create a task.
150			Parameters
151			----------
152			scheduler: :class:`~pincer.utils.tasks.TaskScheduler`
153			The scheduler to use.
154			coro: :class:`~pincer.utils.types.Coro`
155			The coroutine to register as a task.
156			delay: :class:`float`
157			Delay between each iteration of the task.
158			"""
159
160			def __init__(self, scheduler: TaskScheduler, coro: Coro, delay: float):
161			self._scheduler = scheduler
162			self.coro = coro
163			self.delay = delay
164			self._handle: Optional[TimerHandle] = None
165			self._client_required = should_pass_cls(coro)
166
167			def __del__(self):
168			if self.running:
169			self.cancel()
170			else:
171			# Did the user forgot to call task.start() ?
172			_log.warning(
173			"Task `%s` was not scheduled. Did you forget to start it ?",
174			self.coro.__name__
175			)
176
177			@property
178			def cancelled(self):
179			""":class:`bool`: Check if the task has been cancelled or not."""
180			return self.running and self._handle.cancelled()
181
182			@property
183			def running(self):
184			""":class:`bool`: Check if the task is running."""
185			return self._handle is not None
186
187			def start(self):
188			"""Register the task in the TaskScheduler and start
189			the execution of the task.
190			"""
191			if self.running:
192			raise TaskAlreadyRunning(
193			f'Task `{self.coro.__name__}` is already running.', self
194			)
195
196			self._scheduler.register(self)
197
198			def cancel(self):
199			"""Cancel the task."""
200			if not self.running:
201			raise TaskCancelError(
202			f'Task `{self.coro.__name__}` is not running.', self
203			)
204
205			self._handle.cancel()
206			if self in self._scheduler.tasks:
207			self._scheduler.tasks.remove(self)
208

Pincer-org / Pincer

Push — main ( e04df9...e9b147 )

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like