labthings_fastapi.invocation_contexts

Invocation-specific resources provided via context.

This module provides key resources to code that runs as part of an action, specifically a mechanism to allow cancellation, and a way to manage logging. These replace the old dependencies CancelHook and InvocationLogger.

If you are writing action code and want to use logging or allow cancellation, most of the time you should just use get_invocation_logger or cancellable_sleep which are exposed as part of the top-level module.

This module includes lower-level functions that are useful for testing or managing concurrency. Many of these accept an id argument, which is optional. If it is not supplied, we will use the context variables to find the current invocation ID.

Attributes

invocation_id_ctx

Context variable storing the current invocation ID.

Classes

`CancelEvent`	An Event subclass that enables cancellation of actions.
`ThreadWithInvocationID`	A thread that sets a new invocation ID.

Functions

`get_invocation_id`(→ uuid.UUID)	Return the current InvocationID.
`set_invocation_id`(→ collections.abc.Iterator[None])	Set the invocation ID associated with the current context.
`fake_invocation_context`(...)	Set a dummy invocation ID for a block of code.
`get_cancel_event`(→ CancelEvent)	Obtain an event that permits actions to be cancelled.
`cancellable_sleep`(→ None)	Sleep for a specified time, allowing cancellation.
`raise_if_cancelled`(→ None)	Raise an exception if the current invocation has been cancelled.

Module Contents

labthings_fastapi.invocation_contexts.invocation_id_ctx

Context variable storing the current invocation ID.

Note that it is best not to access this directly. Using set_invocation_id is safer, as it ensures proper clean-up and continuity of the cancel event associated with the invocation.

labthings_fastapi.invocation_contexts.get_invocation_id() → uuid.UUID

Return the current InvocationID.

This function returns the ID of the current invocation. This is determined from execution context: it will only succeed if it is called from an action thread.

If this function is called outside of an action thread, it will raise an error.

Returns:: the invocation ID of the current invocation.
Raises:: NoInvocationContextError – if called outside of an action thread.

labthings_fastapi.invocation_contexts.set_invocation_id(id: uuid.UUID) → collections.abc.Iterator[None]

Set the invocation ID associated with the current context.

This is the preferred way to create a new invocation context. As well as setting and cleaning up the invocation ID context variable, this context manager ensures that the cancellation event persists and is not accidentally reset because it’s gone out of scope.

Parameters:: id – The invocation ID to save in the context variable.

labthings_fastapi.invocation_contexts.fake_invocation_context() → collections.abc.Iterator[uuid.UUID]

Set a dummy invocation ID for a block of code.

This function should be used in a with: block.

Yields:: the created invocation ID.

class labthings_fastapi.invocation_contexts.CancelEvent(id: uuid.UUID)

Bases: threading.Event

An Event subclass that enables cancellation of actions.

This threading.Event subclass adds methods to raise InvocationCancelledError exceptions if the invocation is cancelled, usually by a DELETE request to the invocation’s URL.

Initialise a cancellation event.

Only one CancelEvent should exist per invocation. Trying to create a second will raise an error. To avoid this, please use CancelEvent.get_for_id instead of the constructor.

Parameters:: id – The invocation ID.
Raises:: RuntimeError – if a CancelEvent has already been created for the specified invocation ID.

_cancel_events: weakref.WeakValueDictionary[uuid.UUID, Self]: This class-level dictionary ensures only one event exists per invocation ID

invocation_id

classmethod get_for_id(id: uuid.UUID) → Self

Obtain the CancelEvent for a particular Invocation ID.

This is a safe way to obtain an instance of this class, though the top-level function get_cancel_event is recommended.

Only one CancelEvent should exist per Invocation. This method will either create one, or return the existing one.

Parameters:: id – The invocation ID.
Returns:: the cancel event for the given id .

raise_if_set() → None

Raise an exception if the event is set.

An exception will be raised if the event has been set. Before raising the exception, we clear the event. This means that setting the event should raise exactly one exception, and that handling the exception should result in the action continuing to run.

This is intended as a compact alternative to:

if cancel_event.is_set():
    cancel_event.clear()
    raise InvocationCancelledError()

Raises:: InvocationCancelledError – if the event has been cancelled.

sleep(timeout: float) → None

Sleep for a given time in seconds, but raise an exception if cancelled.

This function can be used in place of time.sleep. It will usually behave the same as time.sleep, but if the cancel event is set during the time when we are sleeping, an exception is raised to interrupt the sleep and cancel the action. The event is cleared before raising the exception. This means that handling the exception is sufficient to allow the action to continue.

Parameters:: timeout – The time to sleep for, in seconds.
Raises:: InvocationCancelledError – if the event has been cancelled.

labthings_fastapi.invocation_contexts.get_cancel_event(id: uuid.UUID | None = None) → CancelEvent

Obtain an event that permits actions to be cancelled.

Parameters:: id – The invocation ID. This will be determined from context if not supplied.
Returns:: an event that allows the current invocation to be cancelled.

labthings_fastapi.invocation_contexts.cancellable_sleep(interval: float) → None

Sleep for a specified time, allowing cancellation.

This function should be called from action functions instead of time.sleep to allow them to be cancelled. Usually, this function is equivalent to time.sleep (it waits the specified number of seconds). If the action is cancelled during the sleep, it will raise an InvocationCancelledError to signal that the action should finish.

Warning

This function uses Event.wait internally, which suffers from timing errors on some platforms: it may have error of around 10-20ms. If that’s a problem, consider using time.sleep instead. lt.raise_if_cancelled() may then be used to allow cancellation.

If this function is called from outside of an action thread, it will revert to time.sleep.

Parameters:: interval – The length of time to wait for, in seconds.

labthings_fastapi.invocation_contexts.raise_if_cancelled() → None

Raise an exception if the current invocation has been cancelled.

This function checks for cancellation events and, if the current action invocation has been cancelled, it will raise an InvocationCancelledError to signal the thread to terminate. It is equivalent to cancellable_sleep but without waiting any time.

If called outside of an invocation context, this function does nothing, and will not raise an error.

class labthings_fastapi.invocation_contexts.ThreadWithInvocationID(target: Callable, args: collections.abc.Sequence[Any] | None = None, kwargs: collections.abc.Mapping[str, Any] | None = None, *super_args: Any, **super_kwargs: Any)

Bases: threading.Thread

A thread that sets a new invocation ID.

This is a subclass of threading.Thread and works very much the same way. It implements its functionality by overriding the run method, so this should not be overridden again - you should instead specify the code to run using the target argument.

This function enables an action to be run in a thread, which gets its own invocation ID and cancel hook. This means logs will not be interleaved with the calling action, and the thread may be cancelled just like an action started over HTTP, by calling its cancel method.

The thread also remembers the return value of the target function in the property result and stores any exception raised in the exception property.

A final LabThings-specific feature is cancellation propagation. If the thread is started from an action that may be cancelled, it may be joined with join_and_propagate_cancel. This is intended to be equivalent to calling join but with the added feature that, if the parent thread is cancelled while waiting for the child thread to join, the child thread will also be cancelled.

Initialise a thread with invocation ID.

Parameters:

target – the function to call in the thread.
args – positional arguments to target.
kwargs – keyword arguments to target.
*super_args – arguments passed to threading.Thread.
**super_kwargs – keyword arguments passed to threading.Thread.

_target

_args = []

_kwargs

_invocation_id: uuid.UUID

_result: Any = None

_exception: BaseException | None = None

_cancel_event

property invocation_id: uuid.UUID: The InvocationID of this thread.

property result: Any: The return value of the target function.

property exception: BaseException | None: The exception raised by the target function, or None.

cancel() → None: Set the cancel event to tell the code to terminate.

join_and_propagate_cancel(poll_interval: float = 0.2) → None

Wait for the thread to finish, and propagate cancellation.

This function wraps threading.Thread.join but periodically checks if the calling thread has been cancelled. If it has, it will cancel the thread, before attempting to join it again.

Note that, if the invocation that calls this function is cancelled while the function is running, the exception will propagate, i.e. you should handle InvocationCancelledError unless you wish your invocation to terminate if it is cancelled.

Parameters:: poll_interval – How often to check for cancellation of the calling thread, in seconds.
Raises:: InvocationCancelledError – if this invocation is cancelled while waiting for the thread to join.

run() → None: Run the target function, with invocation ID set in the context variable.