Merge pull request #4 from pomponchik/develop

0.0.4

.github/workflows/tests_and_coverage.yml

@@ -1,4 +1,4 @@
-name: Tests and coverage
+name: Tests
 
 on:
   push

README.md

@@ -1,4 +1,4 @@
-![logo](https://raw.githubusercontent.com/pomponchik/cantok/develop/docs/assets/logo_2.png)
+![logo](https://raw.githubusercontent.com/pomponchik/cantok/main/docs/assets/logo_2.png)
 
 [![Downloads](https://static.pepy.tech/badge/cantok/month)](https://pepy.tech/project/cantok)
 [![Downloads](https://static.pepy.tech/badge/cantok)](https://pepy.tech/project/cantok)
@@ -134,6 +134,39 @@ print(token.cancelled)  # True
 print(token.keep_on())  # False
 ```
 
+There is another method that is close in meaning to `is_cancelled()` - `check()`. It does nothing if the token is not canceled, or raises an exception if canceled. If the token was canceled by calling the `cancel()` method, a `CancellationError` exception will be raised:
+
+```python
+from cantok import SimpleToken
+
+token = SimpleToken()
+token.check()  # Nothing happens.
+token.cancel()
+token.check()  # cantok.errors.CancellationError: The token has been cancelled.
+```
+
+Otherwise, a special exception inherited from `CancellationError` will be raised:
+
+```python
+from cantok import TimeoutToken
+
+token = TimeoutToken(0)
+token.check()  # cantok.errors.TimeoutCancellationError: The timeout of 0 seconds has expired.
+```
+
+Each token class has its own exception and it can be found in the `exception` attribute of the class:
+
+```python
+from cantok import TimeoutToken, CancellationError
+
+token = TimeoutToken(0)
+
+try:
+    token.check()
+except CancellationError as e:
+    print(type(e) is TimeoutToken.exception)  # True
+```
+
 An unlimited number of other tokens can be embedded in one token as arguments during initialization. Each time checking whether it has been canceled, the token first checks its cancellation rules, and if it has not been canceled itself, then it checks the tokens nested in it. Thus, one cancelled token nested in another non-cancelled token cancels it:
 
 ```python
@@ -150,6 +183,20 @@ print(second_token.cancelled)  # False
 print(third_token.cancelled)  # True
 ```
 
+Each exception object has a `token` attribute indicating the specific token that was canceled. This can be useful in situations where several tokens are nested in one another and you want to find out which one has been canceled:
+
+```python
+from cantok import SimpleToken, TimeoutToken, CancellationError
+
+nested_token = TimeoutToken(0)
+token = SimpleToken(nested_token)
+
+try:
+    token.check()
+except CancellationError as e:
+    print(e.token is nested_token)  # True
+```
+
 In addition, any tokens can be summed up among themselves. The summation operation generates another [`SimpleToken`](#simple-token) that includes the previous 2:
 
 ```python

cantok/__init__.py

@@ -4,5 +4,7 @@
 from cantok.tokens.counter_token import CounterToken  # noqa: F401
 from cantok.tokens.timeout_token import TimeoutToken
 
+from cantok.errors import CancellationError, ConditionCancellationError, CounterCancellationError, TimeoutCancellationError  # noqa: F401
+
 
 TimeOutToken = TimeoutToken

cantok/errors.py

@@ -0,0 +1,15 @@
+from typing import Any
+
+class CancellationError(Exception):
+    def __init__(self, message: str, token: Any):
+        self.token = token
+        super().__init__(message)
+
+class ConditionCancellationError(CancellationError):
+    pass
+
+class CounterCancellationError(CancellationError):
+    pass
+
+class TimeoutCancellationError(CancellationError):
+    pass

cantok/tokens/abstract_token.py

@@ -1,10 +1,29 @@
+from enum import Enum
 from abc import ABC, abstractmethod
+from threading import RLock
+from dataclasses import dataclass
+
+from cantok.errors import CancellationError
+
+
+class CancelCause(Enum):
+    CANCELLED = 1
+    SUPERPOWER = 2
+    NOT_CANCELLED = 3
+
+@dataclass
+class CancellationReport:
+    cause: CancelCause
+    from_token: 'AbstractToken'
 
 
 class AbstractToken(ABC):
+    exception = CancellationError
+
     def __init__(self, *tokens: 'AbstractToken', cancelled=False):
         self.tokens = tokens
         self._cancelled = cancelled
+        self.lock = RLock()
 
     def __repr__(self):
         other_tokens = ', '.join([repr(x) for x in self.tokens])
@@ -46,16 +65,28 @@ def keep_on(self) -> bool:
         return not self.is_cancelled()
 
     def is_cancelled(self) -> bool:
-        if self._cancelled:
-            return True
-
-        elif any(x.is_cancelled_reflect() for x in self.tokens):
-            return True
+        return self.get_report().cause != CancelCause.NOT_CANCELLED
 
+    def get_report(self) -> CancellationReport:
+        if self._cancelled:
+            return CancellationReport(
+                cause=CancelCause.CANCELLED,
+                from_token=self,
+            )
         elif self.superpower():
-            return True
+            return CancellationReport(
+                cause=CancelCause.SUPERPOWER,
+                from_token=self,
+            )
 
-        return False
+        for token in self.tokens:
+            if token.is_cancelled_reflect():
+                return token.get_report()
+
+        return CancellationReport(
+            cause=CancelCause.NOT_CANCELLED,
+            from_token=self,
+        )
 
     def is_cancelled_reflect(self):
         return self.is_cancelled()
@@ -74,3 +105,24 @@ def text_representation_of_superpower(self) -> str:  # pragma: no cover
 
     def text_representation_of_extra_kwargs(self) -> str:
         return ''
+
+    def check(self) -> None:
+        with self.lock:
+            if self.is_cancelled_reflect():
+                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.'

cantok/tokens/condition_token.py

@@ -2,9 +2,12 @@
 from contextlib import suppress
 
 from cantok import AbstractToken
+from cantok.errors import ConditionCancellationError
 
 
 class ConditionToken(AbstractToken):
+    exception = ConditionCancellationError
+
     def __init__(self, function: Callable[[], bool], *tokens: AbstractToken, cancelled: bool = False, suppress_exceptions: bool = True, default: bool = False):
         super().__init__(*tokens, cancelled=cancelled)
         self.function = function
@@ -41,3 +44,6 @@ def text_representation_of_extra_kwargs(self) -> str:
             'default': self.default,
         }
         return  ', '.join([f'{key}={value}' for key, value in extra_kwargs.items()])
+
+    def get_superpower_exception_message(self) -> str:
+        return 'The condition is not met.'

cantok/tokens/counter_token.py

@@ -1,17 +1,18 @@
-from threading import RLock
-
 from cantok import AbstractToken
 from cantok import ConditionToken
+from cantok.errors import CounterCancellationError
 
 
 class CounterToken(ConditionToken):
+    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.')
 
         self.counter = counter
+        self.initial_counter = counter
         self.direct = direct
-        self.lock = RLock()
 
         def function() -> bool:
             with self.lock:
@@ -51,3 +52,6 @@ def text_representation_of_superpower(self) -> str:
 
     def text_representation_of_extra_kwargs(self) -> str:
         return f'direct={self.direct}'
+
+    def get_superpower_exception_message(self) -> str:
+        return f'After {self.initial_counter} attempts, the counter was reset to zero.'

cantok/tokens/simple_token.py

@@ -1,9 +1,15 @@
 from cantok import AbstractToken
+from cantok.errors import CancellationError
 
 
 class SimpleToken(AbstractToken):
+    exception = CancellationError
+
     def superpower(self) -> bool:
         return False
 
     def text_representation_of_superpower(self) -> str:
         return ''
+
+    def get_superpower_exception_message(self) -> str:
+        return 'The token has been cancelled.'

cantok/tokens/timeout_token.py

@@ -4,9 +4,12 @@
 
 from cantok import AbstractToken
 from cantok import ConditionToken
+from cantok.errors import TimeoutCancellationError
 
 
 class TimeoutToken(ConditionToken):
+    exception = TimeoutCancellationError
+
     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.')
@@ -32,3 +35,6 @@ def text_representation_of_superpower(self) -> str:
 
     def text_representation_of_extra_kwargs(self) -> str:
         return f'monotonic={self.monotonic}'
+
+    def get_superpower_exception_message(self) -> str:
+        return f'The timeout of {self.timeout} seconds has expired.'

requirements_dev.txt

@@ -4,3 +4,4 @@ twine==4.0.2
 wheel==0.40.0
 mypy==1.4.1
 ruff==0.0.290
+mkdocs-material==9.2.7