stdlibx.cancel package

exception stdlibx.cancel.CancellationTokenCancelledError[source]

Bases: CancellationTokenError

Raised when an operation is cancelled via a cancellation token. This exception indicates that the token was explicitly cancelled.

exception stdlibx.cancel.CancellationTokenError[source]

Bases: Exception

Base class for all cancellation-related errors. Raised when a cancellation event affects the current operation. Serves as the common superclass for all token error types.

exception stdlibx.cancel.CancellationTokenTimeoutError[source]

Bases: CancellationTokenError

Raised when a token is cancelled due to a timeout. This exception indicates that the token’s cancellation was triggered by an internal timeout mechanism (for example via stdlibx.cancel.with_timeout()).

class stdlibx.cancel.CancellationToken(*args, **kwargs)[source]

Bases: Protocol

A cooperative cancellation token interface. Represents a handle that can be checked or awaited to respond to cancellation events triggered elsewhere

get_error()[source]

Get the cancellation error if the token has been cancelled.

Returns:

The Exception if cancelled, or None.

Return type:

Exception | None

is_cancelled()[source]

Check whether the token has been cancelled.

Returns:

True if the token has been cancelled, False otherwise.

Return type:

bool

raise_if_cancelled()[source]

Raise the cancellation error if the token has been cancelled.

Raises:

Exception – If the token is cancelled.

register(fn)[source]

Register a callback to be invoked when the token is cancelled.

Parameters:

fn (Callable[[Exception], None]) – A function to call when cancellation occurs. Receives a Exception.

Return type:

None

wait(timeout)[source]

Block until the token is cancelled or the timeout expires.

Parameters:

timeout (float | None) – Maximum time to wait in seconds, or None to wait indefinitely.

Returns:

The Exception if cancelled, or None if the timeout expires.

Return type:

Exception | None

class stdlibx.cancel.CancelledToken(*args, **kwargs)[source]

Bases: CancellationToken, Protocol

A token that is already cancelled. This specialization of Token ensures that cancellation has occurred and provides non-optional return types

Example:

1if is_token_cancelled(token):
2    reveal_type(token)  # CancelledToken
get_error()[source]

Get the cancellation error.

Returns:

The Exception associated with this cancelled token.

Return type:

Exception

is_cancelled()[source]

Always returns True, since the token is cancelled.

Returns:

True

Return type:

Literal[True]

raise_if_cancelled()[source]

Always raises the cancellation error.

Raises:

Exception – The associated cancellation error.

Return type:

NoReturn

wait(timeout)[source]

Immediately returns the cancellation error.

Parameters:

timeout (float | None) – Ignored. Present for interface compatibility.

Returns:

The Exception associated with this cancelled token.

Return type:

Exception

stdlibx.cancel.default_token()[source]

Create a default, non-cancellable token. This token will never be cancelled and can be used as a placeholder when no cancellation logic is required.

Returns:

A default Token that is never cancelled.

Return type:

CancellationToken

stdlibx.cancel.is_token_cancelled(token)[source]

Type guard to check if a token is cancelled. This function refines the type of the given token to CancelledToken for type checkers if the token is indeed cancelled.

Parameters:

token (CancellationToken) – The token to check.

Returns:

True if the token is cancelled, False otherwise.

Return type:

TypeGuard[CancelledToken]

stdlibx.cancel.with_cancel(parent)[source]

Create a new cancellable token linked to a parent token. The returned token can be cancelled independently, and will also be cancelled if the parent token is cancelled.

Parameters:

parent (CancellationToken) – The parent Token to link cancellation from.

Returns:

A tuple containing, the new Token and a callable to cancel the token manually

Return type:

tuple[CancellationToken, Callable[[], None]]

stdlibx.cancel.with_timeout(parent, timeout)[source]

Create a new cancellable token with a timeout, linked to a parent token. The token will be cancelled if either: the parent token is cancelled or the specified timeout elapses.

Parameters:
  • parent (CancellationToken) – The parent Token to link cancellation from.

  • timeout (float) – Timeout duration in seconds before automatic cancellation.

Returns:

A tuple containing, the new Token and a callable to cancel the token manually

Return type:

tuple[CancellationToken, Callable[[], None]]