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.

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.

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, typing_extensions.Self]: This class-level dictionary ensures only one event exists per invocation ID

invocation_id

classmethod get_for_id(id: uuid.UUID) → typing_extensions.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.