labthings_fastapi.invocation_contexts ===================================== .. py:module:: labthings_fastapi.invocation_contexts .. autoapi-nested-parse:: 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 ---------- .. autoapisummary:: labthings_fastapi.invocation_contexts.invocation_id_ctx Classes ------- .. autoapisummary:: labthings_fastapi.invocation_contexts.CancelEvent Functions --------- .. autoapisummary:: labthings_fastapi.invocation_contexts.get_invocation_id labthings_fastapi.invocation_contexts.set_invocation_id labthings_fastapi.invocation_contexts.fake_invocation_context labthings_fastapi.invocation_contexts.get_cancel_event Module Contents --------------- .. py:data:: 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. .. py:function:: 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. :return: the invocation ID of the current invocation. :raises NoInvocationContextError: if called outside of an action thread. .. py:function:: 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. :param id: The invocation ID to save in the context variable. .. py:function:: 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. .. py:class:: CancelEvent(id: uuid.UUID) Bases: :py:obj:`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. :param id: The invocation ID. :raises RuntimeError: if a `.CancelEvent` has already been created for the specified invocation ID. .. py:attribute:: _cancel_events :type: weakref.WeakValueDictionary[uuid.UUID, typing_extensions.Self] This class-level dictionary ensures only one event exists per invocation ID .. py:attribute:: invocation_id .. py:method:: get_for_id(id: uuid.UUID) -> typing_extensions.Self :classmethod: 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. :param id: The invocation ID. :return: the cancel event for the given ``id`` . .. py:method:: 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: .. code-block:: if cancel_event.is_set(): cancel_event.clear() raise InvocationCancelledError() :raise InvocationCancelledError: if the event has been cancelled. .. py:method:: 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. :param timeout: The time to sleep for, in seconds. :raise InvocationCancelledError: if the event has been cancelled. .. py:function:: get_cancel_event(id: uuid.UUID | None = None) -> CancelEvent Obtain an event that permits actions to be cancelled. :param id: The invocation ID. This will be determined from context if not supplied. :return: an event that allows the current invocation to be cancelled.