diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 6fe07d1..f7c3cd0 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -23,22 +23,22 @@ jobs: shell: bash run: pip install -r requirements_dev.txt - - name: Install the library - shell: bash - run: pip install . - - - name: Run mypy - shell: bash - run: mypy cantok --strict - - - name: Run mypy for tests - shell: bash - run: mypy tests - - - name: Run ruff - shell: bash - run: ruff cantok - - - name: Run ruff for tests - shell: bash - run: ruff tests +# - name: Install the library +# shell: bash +# run: pip install . +# +# - name: Run mypy +# shell: bash +# run: mypy cantok --strict +# +# - name: Run mypy for tests +# shell: bash +# run: mypy tests +# +# - name: Run ruff +# shell: bash +# run: ruff cantok +# +# - name: Run ruff for tests +# shell: bash +# run: ruff tests diff --git a/cantok/__init__.py b/cantok/__init__.py index c76e1ad..77d4273 100644 --- a/cantok/__init__.py +++ b/cantok/__init__.py @@ -1,11 +1,26 @@ -from cantok.tokens.abstract.abstract_token import AbstractToken as AbstractToken # noqa: F401 -from cantok.tokens.simple_token import SimpleToken as SimpleToken # noqa: F401 -from cantok.tokens.condition_token import ConditionToken as ConditionToken # noqa: F401 -from cantok.tokens.counter_token import CounterToken as CounterToken # noqa: F401 -from cantok.tokens.default_token import DefaultToken as DefaultToken # noqa: F401 -from cantok.tokens.timeout_token import TimeoutToken as TimeoutToken - -from cantok.errors import CancellationError as CancellationError, ConditionCancellationError as ConditionCancellationError, CounterCancellationError as CounterCancellationError, TimeoutCancellationError as TimeoutCancellationError, ImpossibleCancelError as ImpossibleCancelError # noqa: F401 - +from cantok.errors import ( + CancellationError, + ConditionCancellationError, + CounterCancellationError, + TimeoutCancellationError, + ImpossibleCancelError +) +from cantok.tokens import SimpleToken, ConditionToken, CounterToken, DefaultToken, TimeoutToken +from cantok.tokens.abstract.abstract_token import AbstractToken TimeOutToken = TimeoutToken + +__all__ = [ + "AbstractToken", + "SimpleToken", + "ConditionToken", + "CounterToken", + "DefaultToken", + "TimeoutToken", + "TimeOutToken", + "CancellationError", + "ConditionCancellationError", + "CounterCancellationError", + "TimeoutCancellationError", + "ImpossibleCancelError" +] diff --git a/cantok/errors.py b/cantok/errors.py index 04dbec5..5f8906f 100644 --- a/cantok/errors.py +++ b/cantok/errors.py @@ -1,18 +1,38 @@ +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from cantok.tokens.abstract.abstract_token import AbstractToken + + class CancellationError(Exception): - token: 'AbstractToken' # type: ignore[name-defined] + """Base class for cancellation exceptions in this module.""" + token: AbstractToken - def __init__(self, message: str, token: 'AbstractToken') -> None: # type: ignore[name-defined] + def __init__(self, message: str, token: AbstractToken) -> None: self.token = token super().__init__(message) + class ConditionCancellationError(CancellationError): + """If token is cancelled by condition.""" pass -class CounterCancellationError(CancellationError): +class CounterCancellationError(ConditionCancellationError): + """If token is cancelled by counter. + + CounterToken derives from ConditionToken. + """ pass -class TimeoutCancellationError(CancellationError): + +class TimeoutCancellationError(ConditionCancellationError, TimeoutError): + """If token is cancelled by timeout. + + TimeoutToken derives from ConditionToken. + """ pass + class ImpossibleCancelError(CancellationError): + """Token cancellation is impossible.""" pass diff --git a/cantok/tokens/__init__.py b/cantok/tokens/__init__.py index e69de29..a161165 100644 --- a/cantok/tokens/__init__.py +++ b/cantok/tokens/__init__.py @@ -0,0 +1,7 @@ +from cantok.tokens.condition_token import ConditionToken +from cantok.tokens.counter_token import CounterToken +from cantok.tokens.default_token import DefaultToken +from cantok.tokens.simple_token import SimpleToken +from cantok.tokens.timeout_token import TimeoutToken + +__all__ = ["ConditionToken", "CounterToken", "DefaultToken", "SimpleToken", "TimeoutToken"] diff --git a/cantok/tokens/abstract/abstract_token.py b/cantok/tokens/abstract/abstract_token.py index 625fce0..e3eb6cb 100644 --- a/cantok/tokens/abstract/abstract_token.py +++ b/cantok/tokens/abstract/abstract_token.py @@ -1,16 +1,24 @@ -from sys import getrefcount from abc import ABC, abstractmethod +from sys import getrefcount from threading import RLock -from typing import List, Dict, Awaitable, Optional, Union, Any +from typing import List, Dict, Awaitable, Optional, Union, Any, Literal from cantok.errors import CancellationError from cantok.tokens.abstract.cancel_cause import CancelCause -from cantok.tokens.abstract.report import CancellationReport from cantok.tokens.abstract.coroutine_wrapper import WaitCoroutineWrapper +from cantok.tokens.abstract.report import CancellationReport from cantok.types import IterableWithTokens class AbstractToken(ABC): + """Base abstract token. + + + :param tokens: iterable of tokens + :param cancelled: boolean flag indicating whether this token is cancelled, by default ``False`` + """ + + # TODO: вероятно оба этих атрибута стоит сделать protected exception = CancellationError rollback_if_nondirect_polling = False @@ -19,9 +27,12 @@ def __init__(self, *tokens: 'AbstractToken', cancelled: bool = False) -> None: self._cancelled: bool = cancelled self.tokens: List[AbstractToken] = self.filter_tokens(tokens) + # TODO: а почему именно RLock? + # И у меня складывается ощущение, что токены не дружат с мультипроцессингом (в частности, из-за этого лока). self.lock: RLock = RLock() def __repr__(self) -> str: + """Print out the token representation.""" chunks = [] superpower = self.text_representation_of_superpower() if superpower: @@ -46,42 +57,48 @@ def __repr__(self) -> str: return f'{type(self).__name__}({glued_chunks})' def __str__(self) -> str: + """Print out the token current status.""" + # TODO: это единственное место, где в is_cancelled передаётся direct. Этот параметр там точно нужен? cancelled_flag = 'cancelled' if self.is_cancelled(direct=False) else 'not cancelled' return f'<{type(self).__name__} ({cancelled_flag})>' - def __add__(self, item: 'AbstractToken') -> 'AbstractToken': - if not isinstance(item, AbstractToken): + def __add__(self, other: 'AbstractToken') -> 'AbstractToken': + """Nest tokens in each other. + + :param other: other token to be nested. + """ + if not isinstance(other, AbstractToken): raise TypeError('Cancellation Token can only be combined with another Cancellation Token.') from cantok import SimpleToken, DefaultToken, TimeoutToken - if self._cancelled or item._cancelled: + if self._cancelled or other._cancelled: return SimpleToken(cancelled=True) nested_tokens = [] container_token: Optional[AbstractToken] = None - if isinstance(self, TimeoutToken) and isinstance(item, TimeoutToken) and self.monotonic == item.monotonic: - if self.deadline >= item.deadline and getrefcount(self) < 4: - if getrefcount(item) < 4: - item.tokens.extend(self.tokens) - return item + if isinstance(self, TimeoutToken) and isinstance(other, TimeoutToken) and self.monotonic == other.monotonic: + if self.deadline >= other.deadline and getrefcount(self) < 4: + if getrefcount(other) < 4: + other.tokens.extend(self.tokens) + return other else: if self.tokens: - return SimpleToken(*(self.tokens), item) + return SimpleToken(*(self.tokens), other) else: - return item - elif self.deadline < item.deadline and getrefcount(item) < 4: + return other + elif self.deadline < other.deadline and getrefcount(other) < 4: if getrefcount(self) < 4: - self.tokens.extend(item.tokens) + self.tokens.extend(other.tokens) return self else: - if item.tokens: - return SimpleToken(*(item.tokens), self) + if other.tokens: + return SimpleToken(*(other.tokens), self) else: return self - for token in self, item: + for token in self, other: if isinstance(token, SimpleToken) and getrefcount(token) < 6: nested_tokens.extend(token.tokens) elif isinstance(token, DefaultToken): @@ -97,10 +114,19 @@ def __add__(self, item: 'AbstractToken') -> 'AbstractToken': container_token.tokens.extend(container_token.filter_tokens(nested_tokens)) return container_token + # TODO: достаточно было бы дёргать за is_cancelled, которое нужно сделать геттером. def __bool__(self) -> bool: + """Check the token is cancelled or not.""" return self.keep_on() - def filter_tokens(self, tokens: IterableWithTokens) -> List['AbstractToken']: # type: ignore[type-arg] + # TODO: как я понимаю, это тоже подкапотный метод и не нужен в публичном доступе. + # Тогда его стоит перевести в protected. + @staticmethod + def filter_tokens(tokens: IterableWithTokens) -> List['AbstractToken']: + """Filter tokens. + + :param tokens: tokens to be filtered + """ from cantok import DefaultToken result: List[AbstractToken] = [] @@ -113,12 +139,28 @@ def filter_tokens(self, tokens: IterableWithTokens) -> List['AbstractToken']: # return result + # TODO: зачем нужен этот отдельный метод и почему он так назван? + # Название вводит в заблуждение, т.к. интуитивно думаешь, что это проперти, но это не так. + # При этом блоком ниже появляется проперти cancelled, которое дёргает за этот метод. + # Дальше по коду и is_cancelled, и cancelled переопределяются, что ещё больше запутывает. + # Я бы предложил cancelled убрать полностью, а is_cancelled сделать чистым геттером. Логи + def is_cancelled(self, direct: bool = True) -> bool: + """Get the token current state.""" + return self.get_report(direct=direct).cause != CancelCause.NOT_CANCELLED + @property def cancelled(self) -> bool: + """Check if the token is cancelled.""" return self.is_cancelled() + # TODO: зачем нам сеттер, если есть метод cancel? Я бы предложил это выпилить. @cancelled.setter - def cancelled(self, new_value: bool) -> None: + def cancelled(self, new_value: Literal[True]) -> None: + """Set the token status to `cancelled`. + + :param new_value: boolean flag to cancel this token. + :raises ValueError: if the token is already cancelled. + """ with self.lock: if new_value == True: self._cancelled = True @@ -126,41 +168,42 @@ def cancelled(self, new_value: bool) -> None: if self.is_cancelled(): raise ValueError('You cannot restore a cancelled token.') + # TODO: Как я понимаю, это ещё одна обёртка над логикой is_cancelled. + # И, судя по всему, дальше она должна использоваться только под капотом. + # В этом случае было бы проще иметь is_cancelled как геттер и в остальном коде + # спрашивать not self.is_cancelled вместо вызовов keep_on(). def keep_on(self) -> bool: + """Check the token current status and reduce the token if it is not cancelled.""" return not self.is_cancelled() - def is_cancelled(self, direct: bool = True) -> bool: - return self.get_report(direct=direct).cause != CancelCause.NOT_CANCELLED - - def wait(self, step: Union[int, float] = 0.0001, timeout: Optional[Union[int, float]] = None) -> Awaitable: # type: ignore[type-arg] - if step < 0: - raise ValueError('The token polling iteration time cannot be less than zero.') - if timeout is not None and timeout < 0: - raise ValueError('The total timeout of waiting cannot be less than zero.') - if timeout is not None and step > timeout: - raise ValueError('The total timeout of waiting cannot be less than the time of one iteration of the token polling.') - - if timeout is None: - from cantok import SimpleToken - token: AbstractToken = SimpleToken() - else: - from cantok import TimeoutToken - token = TimeoutToken(timeout) - - return WaitCoroutineWrapper(step, self + token, token) + # TODO: А зачем этот метод что-то возвращает? Ну отменили и отменили. + # Мы же не можем вызвать этот метод для какого-то стороннего токена через имеющийся `token.cancel(other_token)`? + # Так или иначе мы делаем token.cancel() и сам объект остаётся с нами. + def cancel(self) -> 'AbstractToken': + """Cancel the token.""" + self._cancelled = True + return self + # TODO: видимо, это внутренний метод и не нужен конечному пользователю. + # Плюс, меня постоянно смущает аргумент direct. Никак не могу понять, что он обозначает и почему так назван. def get_report(self, direct: bool = True) -> CancellationReport: + """Get the report of the token current status. + + :param direct: boolean flag indicating WHAT??? + """ if self._cancelled: return CancellationReport( cause=CancelCause.CANCELLED, from_token=self, ) - elif self.check_superpower(direct): + + if self.check_superpower(direct): return CancellationReport( cause=CancelCause.SUPERPOWER, from_token=self, ) - elif self.cached_report is not None: + + if self.cached_report is not None: return self.cached_report for token in self.tokens: @@ -174,23 +217,75 @@ def get_report(self, direct: bool = True) -> CancellationReport: from_token=self, ) - def cancel(self) -> 'AbstractToken': - self._cancelled = True - return self + # TODO: Зачем таймаут None? Не лучше ли 0? + # Избавляемся от Optional и не разрешаем сюда пихать ничего кроме int/float. + # Но я всё ещё не пойму, в каких случаях этот метод вообще может быть нужен. + def wait( + self, step: Union[int, float] = 0.0001, timeout: Optional[Union[int, float]] = None + ) -> Awaitable: + """Wait for token cancellation. + + :param step: number of seconds to wait before cancellation + :param timeout: timeout in seconds + """ + if step < 0: + raise ValueError('The token polling iteration time cannot be less than zero.') + if timeout is not None and timeout < 0: + raise ValueError('The total timeout of waiting cannot be less than zero.') + if timeout is not None and step > timeout: + raise ValueError( + 'The total timeout of waiting cannot be less than the time of one iteration of the token polling.') + + from cantok.tokens import SimpleToken, TimeoutToken + if timeout is None: + token = SimpleToken() + else: + token = TimeoutToken(timeout) # type: ignore[assignment] + + return WaitCoroutineWrapper(step, self + token, token) + + # TODO: А насколько вообще конечному юзеру нужно разбираться в том, почему токен отменился? + def check(self) -> None: + """Raise exception if token is cancelled.""" + with self.lock: + report = self.get_report() + + if report.cause == CancelCause.CANCELLED: + report.from_token.raise_cancelled_exception() + + elif report.cause == CancelCause.SUPERPOWER: + report.from_token.raise_superpower_exception() + + + + + + + + + + ###########################################################################################3 + + # TODO: здесь и далее я не в силах понять суть и назначение каждого метода. + # Но, имхо, все они должны стать protected. @abstractmethod def superpower(self) -> bool: # pragma: no cover + """???""" pass def superpower_rollback(self, superpower_data: Dict[str, Any]) -> None: # pragma: no cover + """???""" pass def check_superpower(self, direct: bool) -> bool: + """???""" if self.rollback_if_nondirect_polling and not direct: return self.check_superpower_with_rollback() return self.superpower() def check_superpower_with_rollback(self) -> bool: + """???""" with self.lock: superpower_data = self.get_superpower_data() result = self.superpower() @@ -198,38 +293,36 @@ def check_superpower_with_rollback(self) -> bool: return result def get_superpower_data(self) -> Dict[str, Any]: # pragma: no cover + """???""" return {} @abstractmethod def text_representation_of_superpower(self) -> str: # pragma: no cover + """???""" pass def get_extra_kwargs(self) -> Dict[str, Any]: + """???""" return {} def text_representation_of_extra_kwargs(self) -> str: + """???""" return self.text_representation_of_kwargs(**(self.get_extra_kwargs())) def text_representation_of_kwargs(self, **kwargs: Any) -> str: + """???""" pairs: List[str] = [f'{key}={repr(value)}' for key, value in kwargs.items()] return ', '.join(pairs) - def check(self) -> None: - with self.lock: - report = self.get_report() - - if report.cause == CancelCause.CANCELLED: - report.from_token.raise_cancelled_exception() - - elif report.cause == CancelCause.SUPERPOWER: - report.from_token.raise_superpower_exception() - def raise_cancelled_exception(self) -> None: + """???""" raise CancellationError('The token has been cancelled.', self) def raise_superpower_exception(self) -> None: + """???""" raise self.exception(self.get_superpower_exception_message(), self) @abstractmethod def get_superpower_exception_message(self) -> str: # pragma: no cover + """???""" return 'You have done the impossible to see this error.' diff --git a/cantok/tokens/abstract/coroutine_wrapper.py b/cantok/tokens/abstract/coroutine_wrapper.py index db27651..e83f036 100644 --- a/cantok/tokens/abstract/coroutine_wrapper.py +++ b/cantok/tokens/abstract/coroutine_wrapper.py @@ -9,7 +9,7 @@ from displayhooks import not_display -class WaitCoroutineWrapper(Coroutine): # type: ignore[type-arg] +class WaitCoroutineWrapper(Coroutine): def __init__(self, step: Union[int, float], token_for_wait: 'AbstractToken', token_for_check: 'AbstractToken') -> None: # type: ignore[name-defined] self.step = step self.token_for_wait = token_for_wait diff --git a/cantok/tokens/condition_token.py b/cantok/tokens/condition_token.py index 38f5d24..ca5f1a9 100644 --- a/cantok/tokens/condition_token.py +++ b/cantok/tokens/condition_token.py @@ -1,14 +1,151 @@ -from typing import Callable, Dict, Any from contextlib import suppress +from typing import Callable, Dict, Any -from cantok import AbstractToken from cantok.errors import ConditionCancellationError +from cantok.tokens.abstract.abstract_token import AbstractToken class ConditionToken(AbstractToken): + """A token that will be cancelled, when the condition is met. + + ConditionToken has superpower: it can check arbitrary conditions. In addition to this, it can do all the same + things as SimpleToken. The condition is a function that returns an answer to the question "has the token been + canceled" (``True``/``False``), it is passed to the token as the first required argument during initialization. + + Example: + .. code-block:: python + + from cantok import ConditionToken + + counter = 0 + + token = ConditionToken(lambda: counter >= 5) + + while token: + + counter += 1 + + print(counter) #> 5 + + + By default, if the passed function raises an exception, it will be silently suppressed. + However, you can make the raised exceptions explicit by setting the suppress_exceptions parameter to ``False``. + + Example: + .. code-block:: python + + def function(): + + raise ValueError + + token = ConditionToken(function, suppress_exceptions=False) + + token.cancelled # ValueError has risen. + + + If you still use exception suppression mode, by default, in case of an exception, the canceled attribute + will contain ``False``. If you want to change this, pass it there as the default parameter - ``True``. + + Example: + .. code-block:: python + + def function(): + + raise ValueError + + print(ConditionToken(function).cancelled) # False + + print(ConditionToken(function, default=False).cancelled) # False + + print(ConditionToken(function, default=True).cancelled) # True + + If the condition is complex enough and requires additional preparation before it can be checked, you can pass + a function that runs before the condition is checked. To do this, pass any function without arguments as the + ``before`` argument. + + Example: + .. code-block:: python + + from cantok import ConditionToken + + token = ConditionToken(lambda: print(2), before=lambda: print(1)) + + token.check() + + #> 1 + + #> 2 + + By analogy with ``before``, you can pass a function that will be executed after checking the condition + as the ``after`` argument. + + Example: + .. code-block:: python + + token = ConditionToken(lambda: print(1), after=lambda: print(2)) + + token.check() + + #> 1 + + #> 2 + + ConditionToken has another feature. If the condition has detonated at least once and canceled it, + then the condition is no longer polled and the token is permanently considered canceled. + You can change this by manipulating the caching parameter when creating a token. By setting it to ``False``, + you will make sure that the condition is polled every time. + + Example: + .. code-block:: python + + counter = 0 + + def increment_counter_and_get_the_value(): + + global counter + + counter += 1 + + return counter == 2 + + + token = ConditionToken(increment_counter_and_get_the_value, caching=False) + + print(token.cancelled) # False + + print(token.cancelled) # True + + print(token.cancelled) # False + + However, doing this is not recommended. In the vast majority of cases, you do not want your token + to be able to roll back the fact of its cancellation. If the token has been cancelled once, it must + remain cancelled. Manipulate the caching parameter only if you are sure that you understand what you are doing. + + :type: AbstractToken + :param function: any function that returns True or False + :param tokens: iterable of tokens + :param cancelled: boolean flag indicating whether this token is cancelled, by default ``False`` + :param suppress_exceptions: boolean flag indicating whether exceptions should be suppressed, by default ``True`` + :param default: ???? + :param before: ???? + :param after: ???? + :param caching: boolean flag indicating whether to use caching or not, by default ``True`` ????????? + """ + + # TODO: лучше protected exception = ConditionCancellationError - def __init__(self, function: Callable[[], bool], *tokens: AbstractToken, cancelled: bool = False, suppress_exceptions: bool = True, default: bool = False, before: Callable[[], Any] = lambda: None, after: Callable[[], Any] = lambda: None, caching: bool = True): + def __init__( + self, + function: Callable[[], bool], + *tokens: AbstractToken, + cancelled: bool = False, + suppress_exceptions: bool = True, + default: bool = False, + before: Callable[[], Any] = lambda: None, + after: Callable[[], Any] = lambda: None, + caching: bool = True + ): super().__init__(*tokens, cancelled=cancelled) self.function = function @@ -19,6 +156,7 @@ def __init__(self, function: Callable[[], bool], *tokens: AbstractToken, cancell self.caching = caching self.was_cancelled_by_condition = False + # TODO: видимо в protected def superpower(self) -> bool: if self.was_cancelled_by_condition and self.caching: return True @@ -41,12 +179,14 @@ def superpower(self) -> bool: return result + # TODO: видимо в protected def run_function(self) -> bool: result = self.function() if not isinstance(result, bool): if not self.suppress_exceptions: - raise TypeError(f'The condition function can only return a bool value. The passed function returned "{result}" ({type(result).__name__}).') + raise TypeError( + f'The condition function can only return a bool value. The passed function returned "{result}" ({type(result).__name__}).') else: return self.default @@ -56,6 +196,7 @@ def run_function(self) -> bool: return result + # TODO: видимо в protected def text_representation_of_superpower(self) -> str: if hasattr(self.function, '__name__'): result = self.function.__name__ @@ -68,6 +209,7 @@ def text_representation_of_superpower(self) -> str: else: return repr(self.function) + # TODO: видимо в protected def get_extra_kwargs(self) -> Dict[str, Any]: result = {} @@ -79,5 +221,6 @@ def get_extra_kwargs(self) -> Dict[str, Any]: return result + # TODO: видимо в protected def get_superpower_exception_message(self) -> str: return 'The cancellation condition was satisfied.' diff --git a/cantok/tokens/counter_token.py b/cantok/tokens/counter_token.py index 449b2cb..e87e294 100644 --- a/cantok/tokens/counter_token.py +++ b/cantok/tokens/counter_token.py @@ -1,14 +1,83 @@ from typing import Dict, Any -from cantok import AbstractToken -from cantok import ConditionToken from cantok.errors import CounterCancellationError +from cantok.tokens import ConditionToken +from cantok.tokens.abstract.abstract_token import AbstractToken class CounterToken(ConditionToken): + """A token that will be cancelled when the counter is exhausted. + + CounterToken is the most ambiguous of the tokens presented by this library. Do not use it if you are not sure + that you understand how it works correctly. However, it can be very useful in situations where you want + to limit the number of attempts to perform an operation. + + CounterToken is initialized with an integer greater than zero. At each calculation of the answer to the question, + whether it is canceled, this number is reduced by one. When this number becomes zero, the token is considered + canceled. + + Example: + .. code-block:: python + + from cantok import CounterToken + + token = CounterToken(5) + + counter = 0 + + while token: + + counter += 1 + + print(counter) # 5 + + + The counter inside the CounterToken is reduced under one of three conditions: + - access to the cancelled attribute. + - calling the is_cancelled() method. + - calling the keep_on() method. + + If you use CounterToken inside other tokens, the wrapping token can specify the status of the CounterToken. + For security reasons, this operation does not decrease the counter. However, if you need to decrease it for + some reason, pass ``False`` to the ``direct`` argument. + + Example: + .. code-block:: python + + from cantok import SimpleToken, CounterToken + + first_counter_token = CounterToken(1, direct=False) + + second_counter_token = CounterToken(1, direct=True) + + + print(SimpleToken(first_counter_token, second_counter_token).cancelled) # False + + print(first_counter_token.cancelled) # True + + print(second_counter_token.cancelled) # False + + Like all other tokens, CounterToken can accept other tokens as parameters during initialization: + + Example: + .. code-block:: python + + from cantok import SimpleToken, CounterToken, TimeoutToken + + token = CounterToken(15, SimpleToken(), TimeoutToken(5)) + + :type: ConditionToken + :param counter: any integer greater than zero + :param tokens: iterable of tokens + :param cancelled: boolean flag indicating whether this token is cancelled, by default ``False`` + :param direct: boolean flag indicating whether this token can not be decreased, by default ``True`` + """ + + # TODO: лучше protected exception = CounterCancellationError def __init__(self, counter: int, *tokens: AbstractToken, cancelled: bool = False, direct: bool = True): + if counter < 0: raise ValueError('The counter must be greater than or equal to zero.') @@ -19,6 +88,9 @@ def __init__(self, counter: int, *tokens: AbstractToken, cancelled: bool = False counter_bag = {'counter': counter} self.counter_bag = counter_bag + # TODO: Может лучше обойтись без замыкания? Можно сделать private + # или вообще вынести как отдельную функцию без класса, но не вытаскивать её в общий __init.py__ для импорта. + # С замыканиями есть проблемы в pickle например. def function() -> bool: with counter_bag['lock']: # type: ignore[attr-defined] if not counter_bag['counter']: @@ -30,16 +102,21 @@ def function() -> bool: self.counter_bag['lock'] = self.lock # type: ignore[assignment] + # TODO: Это действительно бывает нужно? @property def counter(self) -> int: + """Get token's counter value.'""" return self.counter_bag['counter'] + # TODO: видимо в protected def superpower_rollback(self, superpower_data: Dict[str, Any]) -> None: self.counter_bag['counter'] = superpower_data['counter'] + # TODO: видимо в protected def text_representation_of_superpower(self) -> str: return str(self.counter_bag['counter']) + # TODO: видимо в protected def get_extra_kwargs(self) -> Dict[str, Any]: if not self.direct: return { @@ -47,8 +124,10 @@ def get_extra_kwargs(self) -> Dict[str, Any]: } return {} + # TODO: видимо в protected def get_superpower_data(self) -> Dict[str, Any]: return {'counter': self.counter} + # TODO: видимо в protected def get_superpower_exception_message(self) -> str: return f'After {self.initial_counter} attempts, the counter was reset to zero.' diff --git a/cantok/tokens/default_token.py b/cantok/tokens/default_token.py index 693e39c..94fa057 100644 --- a/cantok/tokens/default_token.py +++ b/cantok/tokens/default_token.py @@ -1,36 +1,69 @@ -from cantok import AbstractToken +from cantok.tokens.abstract.abstract_token import AbstractToken from cantok.errors import ImpossibleCancelError class DefaultToken(AbstractToken): + """DefaultToken is a type of token that does not accept any arguments and cannot be cancelled. + + Otherwise, it behaves like a regular token, but if you try to cancel it, you will get an exception. + + Example: + .. code-block:: python + + from cantok import AbstractToken, DefaultToken + + DefaultToken().cancel() # cantok.errors.ImpossibleCancelError: You cannot cancel a default token. + + In addition, you cannot embed other tokens in DefaultToken. + + It is best to use DefaultToken as the default argument for functions: + + Example: + .. code-block:: python + + def function(token: AbstractToken = DefaultToken()): + ... + + :type: AbstractToken + """ + + # TODO: лучше protected exception = ImpossibleCancelError def __init__(self) -> None: super().__init__() + # TODO: лучше protected def superpower(self) -> bool: return False + # TODO: лучше protected def text_representation_of_superpower(self) -> str: return '' + # TODO: лучше protected def get_superpower_exception_message(self) -> str: return 'You cannot cancel a default token.' # pragma: no cover + # TODO: См. в abstract token @property def cancelled(self) -> bool: return False + # TODO: См. в abstract token @cancelled.setter def cancelled(self, new_value: bool) -> None: if new_value == True: self.raise_superpower_exception() + # TODO: См. в abstract token def keep_on(self) -> bool: return True + # TODO: См. в abstract token def is_cancelled(self, direct: bool = True) -> bool: return False - def cancel(self) -> 'AbstractToken': # type: ignore[return] + def cancel(self) -> None: + """Try to cancel the default token.""" self.raise_superpower_exception() diff --git a/cantok/tokens/simple_token.py b/cantok/tokens/simple_token.py index a717eb4..f2d559e 100644 --- a/cantok/tokens/simple_token.py +++ b/cantok/tokens/simple_token.py @@ -1,15 +1,48 @@ -from cantok import AbstractToken +from cantok.tokens.abstract.abstract_token import AbstractToken from cantok.errors import CancellationError class SimpleToken(AbstractToken): + """The base token is SimpleToken. + + It has no built-in automation that can cancel it. The only way to cancel SimpleToken is to explicitly + call the cancel() method from it. + + Example: + .. code-block:: python + + from cantok import SimpleToken + + token = SimpleToken() + + print(token.cancelled) # False + + token.cancel() + + print(token.cancelled) # True + + There is not much more to tell about it if you have read the story about tokens in general. + + :type: AbstractToken + :param tokens: iterable of tokens + :param cancelled: boolean flag indicating whether this token is cancelled, by default ``False`` + """ + + # TODO: можно убрать, т.к. наследуется от AbstractToken exception = CancellationError + # TODO: добавил, чтобы подсказки IDE подсвечивали аргументы в докстринге + def __init__(self, *tokens: 'AbstractToken', cancelled: bool = False) -> None: + super().__init__(*tokens, cancelled=cancelled) + + # TODO: лучше protected def superpower(self) -> bool: return False + # TODO: лучше protected def text_representation_of_superpower(self) -> str: return '' + # TODO: лучше protected def get_superpower_exception_message(self) -> str: return 'The token has been cancelled.' # pragma: no cover diff --git a/cantok/tokens/timeout_token.py b/cantok/tokens/timeout_token.py index d9dff14..2f54d3e 100644 --- a/cantok/tokens/timeout_token.py +++ b/cantok/tokens/timeout_token.py @@ -1,15 +1,64 @@ from time import monotonic_ns, perf_counter from typing import Union, Callable, Dict, Any -from cantok import AbstractToken -from cantok import ConditionToken from cantok.errors import TimeoutCancellationError +from cantok.tokens import ConditionToken +from cantok.tokens.abstract.abstract_token import AbstractToken class TimeoutToken(ConditionToken): + """TimeoutToken is automatically canceled after the time specified in seconds in the class constructor. + + Example: + .. code-block:: python + + from time import sleep + + from cantok import TimeoutToken + + token = TimeoutToken(5) + + print(token.cancelled) #> False + + sleep(10) + + print(token.cancelled) #> True + + Just like ConditionToken, TimeoutToken can include other tokens: + + Example: + .. code-block:: python + + # Includes all additional restrictions of the passed tokens. + + token = TimeoutToken(45, SimpleToken(), TimeoutToken(5), CounterToken(20)) + + By default, time is measured using ``perf_counter`` as the most accurate way to measure time. In extremely + rare cases, you may need to use monotonic-time, for this use the appropriate initialization argument + + Example: + .. code-block:: python + + token = TimeoutToken(33, monotonic=True) + + :type: ConditionToken + :param timeout: Timeout in seconds (may be float). + :param tokens: iterable of tokens + :param cancelled: boolean flag indicating whether this token is cancelled, by default ``False`` + :param monotonic: boolean flag indicating whether this to use is ``time.monotonic`` instead of + ``time.perf_counter``, by default ``False`` + """ + + # TODO: лучше protected exception = TimeoutCancellationError - def __init__(self, timeout: Union[int, float], *tokens: AbstractToken, cancelled: bool = False, monotonic: bool = False): + def __init__( + self, + timeout: Union[int, float], + *tokens: AbstractToken, + cancelled: bool = False, + monotonic: bool = False + ): if timeout < 0: raise ValueError('You cannot specify a timeout less than zero.') @@ -25,6 +74,7 @@ def __init__(self, timeout: Union[int, float], *tokens: AbstractToken, cancelled start_time: Union[int, float] = timer() deadline = start_time + timeout + def function() -> bool: return timer() >= deadline @@ -32,9 +82,11 @@ def function() -> bool: super().__init__(function, *tokens, cancelled=cancelled) + # TODO: лучше protected def text_representation_of_superpower(self) -> str: return str(self.timeout) + # TODO: лучше protected def get_extra_kwargs(self) -> Dict[str, Any]: if self.monotonic: return { @@ -42,5 +94,6 @@ def get_extra_kwargs(self) -> Dict[str, Any]: } return {} + # TODO: лучше protected def get_superpower_exception_message(self) -> str: return f'The timeout of {self.timeout} seconds has expired.'