Implementation of the "Cancellation Token" pattern
Project description
Cancellation Token is a pattern that allows us to refuse to continue calculations that we no longer need. It is implemented out of the box in many programming languages, for example in C# and in Go. However, there was still no sane implementation in Python, until the cantok library appeared.
Table of contents
Quick start
Install it:
pip install cantok
And use:
from random import randint
from cantok import ConditionToken, CounterToken, TimeoutToken
token = ConditionToken(lambda: randint(1, 100_000) == 1984) + CounterToken(400_000, direct=False) + TimeoutToken(1)
counter = 0
while token:
counter += 1
print(counter)
In this example, we use a token that describes several restrictions: on the number of iterations of the cycle, on time, as well as on the occurrence of a random unlikely event. When any of the indicated events occur, the cycle stops.
Read more about the possibilities of tokens, as well as about the pattern in general.
The pattern
The essence of the pattern is that we pass special objects to functions and constructors, by which the executed code can understand whether it should continue its execution or not. When deciding whether to allow code execution to continue, this object can take into account both the restrictions specified to it, such as the maximum code execution time, and receive signals about the need to stop from the outside, for example from another thread or a coroutine. Thus, we do not nail down the logic associated with stopping code execution, for example, by directly tracking cycle counters, but implement Dependency Injection of this restriction.
In addition, the pattern assumes that various restrictions can be combined indefinitely with each other: if at least one of the restrictions is not met, code execution will be interrupted. It is assumed that each function in the call stack will call other functions, throwing its token directly to them, or wrapping it in another token, with a stricter restriction imposed on it.
Unlike other ways of interrupting code execution, tokens do not force the execution thread to be interrupted forcibly. The interruption occurs "gently", allowing the code to terminate correctly, return all occupied resources and restore consistency.
It is highly desirable for library developers to use this pattern for any long-term composite operations. Your function can accept a token as an optional argument, with a default value that imposes minimal restrictions or none at all. If the user wishes, he can transfer his token there, imposing stricter restrictions on the library code. In addition to a more convenient and extensible API, this will give the library an advantage in the form of better testability, because the restrictions are no longer sewn directly into the function, which means they can be made whatever you want for the test. In addition, the library developer no longer needs to think about all the numerous restrictions that can be imposed on his code - the user can take care of it himself if he needs to.
Tokens
All token classes presented in this library have a uniform interface. All tokens are thread-safe. And they are all inherited from one class: AbstractToken
. The only reason why you might want to import it is to use it for a type hint. This example illustrates a type hint suitable for any of the tokens:
from cantok import AbstractToken
def function(token: AbstractToken):
...
Each token object has a cancelled
attribute and a cancel()
method. By the attribute, you can find out whether this token has been canceled:
from cantok import SimpleToken
token = SimpleToken()
print(token.cancelled) # False
token.cancel()
print(token.cancelled) # True
The cancelled attribute is dynamically calculated and takes into account, among other things, specific conditions that are checked by a specific token. Here is an example with a token that measures time:
from time import sleep
from cantok import TimeoutToken
token = TimeoutToken(5)
print(token.cancelled) # False
sleep(10)
print(token.cancelled) # True
In addition to this attribute, each token implements the is_cancelled()
method. It does exactly the same thing as the attribute:
from cantok import SimpleToken
token = SimpleToken()
print(token.cancelled) # False
print(token.is_cancelled()) # False
token.cancel()
print(token.cancelled) # True
print(token.is_cancelled()) # True
Choose what you like best. To the author of the library, the use of the attribute seems more beautiful, but the method call more clearly reflects the complexity of the work that is actually being done to answer the question "has the token been canceled?".
There is another method opposite to is_cancelled()
- keep_on()
. It answers the opposite question, and can be used in the same situations:
from cantok import SimpleToken
token = SimpleToken()
print(token.cancelled) # False
print(token.keep_on()) # True
token.cancel()
print(token.cancelled) # True
print(token.keep_on()) # False
You don't have to call the keep_on()
method directly. Use the token itself as a boolean value, and the method call will occur "under the hood" automatically:
from cantok import SimpleToken
token = SimpleToken()
print(bool(token)) # True
print(token.keep_on()) # True
token.cancel()
print(bool(token)) # False
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:
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:
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:
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:
from cantok import SimpleToken
first_token = SimpleToken()
second_token = SimpleToken()
third_token = SimpleToken(first_token, second_token)
first_token.cancel()
print(first_token.cancelled) # True
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:
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
that includes the previous 2:
from cantok import SimpleToken, TimeoutToken
print(repr(SimpleToken() + TimeoutToken(5)))
# SimpleToken(SimpleToken(), TimeoutToken(5, monotonic=False))
This feature is convenient to use if your function has received a token with certain restrictions and wants to throw it into other called functions, imposing additional restrictions:
from cantok import AbstractToken, TimeoutToken
def function(token: AbstractToken):
...
another_function(token + TimeoutToken(5)) # Imposes an additional restriction on the function being called: work for no more than 5 seconds. At the same time, it does not know anything about what restrictions were imposed earlier.
...
Read on about the features of each type of tokens in more detail.
Simple token
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.
from cantok import SimpleToken
token = SimpleToken()
print(token.cancelled) # False
token.cancel()
print(token.cancelled) # True
SimpleToken
is also implicitly generated by the operation of summing two other tokens:
from cantok import CounterToken, TimeoutToken
print(repr(CounterToken(5) + TimeoutToken(5)))
# SimpleToken(CounterToken(5, direct=True), TimeoutToken(5, monotonic=False))
There is not much more to tell about it if you have read the story about tokens in general.
Condition token
A slightly more complex type of token than SimpleToken
is ConditionToken
. In addition to everything that SimpleToken
does, it also checks the condition passed to it as a first argument, answering the question whether it has been canceled.
To initialize ConditionToken
, pass a function to it that does not accept arguments and returns a boolean value. If it returns True
, it means that the operation has been canceled:
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
:
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
.
def function(): raise ValueError
print(ConditionToken(function).cancelled) # False
print(ConditionToken(function, default=False).cancelled) # False
print(ConditionToken(function, default=True).cancelled) # True
ConditionToken
may include other tokens during initialization:
token = ConditionToken(lambda: False, SimpleToken(), TimeoutToken(5), CounterToken(20)) # Includes all additional restrictions of the passed tokens.
Timeout token
TimeoutToken
is automatically canceled after the time specified in seconds in the class constructor:
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:
token = TimeoutToken(45, SimpleToken(), TimeoutToken(5), CounterToken(20)) # Includes all additional restrictions of the passed tokens.
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:
token = TimeoutToken(33, monotonic=True)
Counter token
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:
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 for some reason you need it to decrease, pass direct
- False
as an argument:
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:
from cantok import SimpleToken, CounterToken, TimeoutToken
token = CounterToken(15, SimpleToken(), TimeoutToken(5))
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.