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 `~lt.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 labthings_fastapi.invocation_contexts.ThreadWithInvocationID 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 labthings_fastapi.invocation_contexts.cancellable_sleep labthings_fastapi.invocation_contexts.raise_if_cancelled 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, 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) -> 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. .. py:function:: 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`\ . :param interval: The length of time to wait for, in seconds. .. py:function:: 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 `~lt.cancellable_sleep` but without waiting any time. If called outside of an invocation context, this function does nothing, and will not raise an error. .. py:class:: 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: :py:obj:`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. :param target: the function to call in the thread. :param args: positional arguments to ``target``\ . :param kwargs: keyword arguments to ``target``\ . :param \*super_args: arguments passed to `threading.Thread`\ . :param \*\*super_kwargs: keyword arguments passed to `threading.Thread`\ . .. py:attribute:: _target .. py:attribute:: _args :value: [] .. py:attribute:: _kwargs .. py:attribute:: _invocation_id :type: uuid.UUID .. py:attribute:: _result :type: Any :value: None .. py:attribute:: _exception :type: BaseException | None :value: None .. py:attribute:: _cancel_event .. py:property:: invocation_id :type: uuid.UUID The InvocationID of this thread. .. py:property:: result :type: Any The return value of the target function. .. py:property:: exception :type: BaseException | None The exception raised by the target function, or None. .. py:method:: cancel() -> None Set the cancel event to tell the code to terminate. .. py:method:: 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. :param 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. .. py:method:: run() -> None Run the target function, with invocation ID set in the context variable.