raise_if_cancelled

raise_if_cancelled#

rapidtools.core.raise_if_cancelled(cancel_event: Event | None, context: str = '') → None[source]#

Raise OperationCancelled if cancel_event has been set.

Components call this between units of work so that a caller running them on a background thread can stop them promptly. Passing None disables cancellation entirely, which keeps the call sites free of if guards.

Parameters:
  • cancel_event (Event | None) – The event to inspect. None means cancellation is not enabled.

  • context (str, optional) – A short description of the work in progress (e.g. 'tile 3/10') that is appended to the exception message. Defaults to ''.

Raises:

OperationCancelled – If cancel_event is not None and 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.