raise_if_cancelled#
- rapidtools.core.raise_if_cancelled(cancel_event: Event | None, context: str = '') None[source]#
Raise
OperationCancelledifcancel_eventhas been set.Components call this between units of work so that a caller running them on a background thread can stop them promptly. Passing
Nonedisables cancellation entirely, which keeps the call sites free ofifguards.- Parameters:
- Raises:
OperationCancelled – If
cancel_eventis notNoneand is set.
Example
>>> import threading >>> from rapidtools.core import raise_if_cancelled >>> >>> stop = threading.Event() >>> raise_if_cancelled(stop) # not set: returns silently >>> raise_if_cancelled(None) # cancellation disabled: returns silently >>> stop.set() >>> raise_if_cancelled(stop, context='inference') Traceback (most recent call last): ... OperationCancelled: Operation cancelled during inference.