labthings_fastapi.actions

Actions module.

Actions are represented by methods, decorated with the lt.action decorator.

See the Actions documentation for a top-level overview of actions in LabThings-FastAPI.

Developer notes

Currently much of the code related to Actions is in lt.action and the underlying ActionDescriptor. This is likely to be refactored in the near future.

Classes

`Invocation`	A Thread subclass that retains output values and tracks progress.
`ActionManager`	A class to manage a collection of actions.

Module Contents

class labthings_fastapi.actions.Invocation(action: ActionDescriptor, thing: labthings_fastapi.thing.Thing, id: uuid.UUID, input: pydantic.BaseModel | None = None, dependencies: dict[str, Any] | None = None, log_len: int = 1000)

Bases: threading.Thread

A Thread subclass that retains output values and tracks progress.

Invocation threads add several bits of functionality compared to the base threading.Thread.

They are instantiated with an ActionDescriptor and a Thing rather than a target function (see __init__).
Each invocation is assigned a unique ID to allow it to be polled over HTTP.
A CancelHook is provided to allow the invocation to stop gracefully if it is cancelled by the user.

Create a thread to run an action and track its outputs.

Parameters:

action – provides the function that we run, as well as metadata and type information. The descriptor is not bound to an object, so we supply the Thing it’s bound to when the function is run.
thing – is the object on which we are running the action, i.e. it is supplied to the function wrapped by action as the self argument.
id – is a uuid.UUID used to identify the invocation, for example when polling its status via HTTP.
input – is a pydantic.BaseModel representing the body of the HTTP request that invoked the action. It is supplied to the function as keyword arguments.
dependencies – is a dictionary of keyword arguments, supplied by FastAPI by its dependency injection mechanism.
log_len – sets the number of log entries that will be held in memory by the invocation’s logger.

action_ref

thing_ref

input

dependencies

_ID

retention_time

expiry_time: datetime.datetime | None = None

_status_lock

_status: labthings_fastapi.invocations.InvocationStatus

_output_model_instance: pydantic.BaseModel | None = None

_request_time: datetime.datetime

_start_time: datetime.datetime | None = None

_end_time: datetime.datetime | None = None

_exception: Exception | None = None

_log: collections.deque

property id: uuid.UUID: UUID for the thread. Note this not the same as the native thread ident.

property output: Any: Return value of the Action. If the Action is still running, returns None.

property output_model_instance: pydantic.BaseModel | None: Return value of the Action, as a model, or None.

property log: list[logging.LogRecord]: A list of log items generated by the Action.

property status: labthings_fastapi.invocations.InvocationStatus

Current running status of the thread.

See InvocationStatus for the values and their meanings.

property thing: labthings_fastapi.thing.Thing

The Thing to which the action is bound, i.e. this is self.

Raises:: RuntimeError – if the Thing no longer exists.

property cancel_hook: labthings_fastapi.invocation_contexts.CancelEvent: The cancel event associated with this Invocation.

cancel() → None

Cancel the task by requesting the code to stop.

This is an opt-in feature: the action must use a CancelHook dependency and periodically check it.

summary_model() → labthings_fastapi.invocations.InvocationSummary

Generate a summary of the invocation suitable for HTTP.

Returns:: an InvocationSummary representing this Invocation.

response() → labthings_fastapi.invocations.InvocationModel

Generate a representation of the invocation suitable for HTTP.

When an invocation is polled, we return a JSON object that includes its status, any log entries, a return value (if completed), and a link to poll for updates.

Returns:: an InvocationModel representing this Invocation.

_publish_status() → None

Publish a status change event to any observers.

This should be called after each change to self._status

run() → None

Run the action and track progress.

Invocation overrides the default threading.Thread.run method to add ways to track its progress and capture the return value.

The code to be run is the function wrapped in the ActionDescriptor that is passed in as action. Its arguments are the associated Thing (the first argument, i.e. self), the input model (split into keyword arguments for each field), and any dependencies (also as keyword arguments).

We update the status of the action by setting self._status and emitting a changed event. This runs async code in the event loop that informs any clients listening over websockets that the event’s status has changed.

Logs are retained by a custom log handler, and are included when the Invocation is serialised over HTTP.

If exceptions are raised by the action code, these are caught and stored. The status is then set to ERROR and the thread terminates.

See Invocation.status for status values.

Raises:: RuntimeError – if there is no Thing associated with the invocation.

class labthings_fastapi.actions.ActionManager

A class to manage a collection of actions.

Set up an ActionManager.

_invocations: dict[uuid.UUID, Invocation]

_invocations_lock

property invocations: list[Invocation]: A list of all the Invocation objects running or recently completed.

append_invocation(invocation: Invocation) → None

Add an Invocation to the ActionManager.

Parameters:: invocation – The Invocation to add.

invoke_action(action: ActionDescriptor, thing: labthings_fastapi.thing.Thing, input: Any, dependencies: dict[str, Any]) → Invocation

Invoke an action, returning the thread where it’s running.

See Invocation for more details.

Parameters:

action – provides the function that we run, as well as metadata and type information. The descriptor is not bound to an object, so we supply the Thing it’s bound to when the function is run.
thing – is the object on which we are running the action, i.e. it is supplied to the function wrapped by action as the self argument.
input – is a pydantic.BaseModel representing the body of the HTTP request that invoked the action. It is supplied to the function as keyword arguments.
dependencies – is a dictionary of keyword arguments, supplied by FastAPI by its dependency injection mechanism.

Returns:

an Invocation object that has been started.

get_invocation(id: uuid.UUID) → Invocation

Retrieve an invocation by ID.

Parameters:: id – the unique ID of the action to retrieve.
Returns:: the Invocation object.

list_invocations(action: ActionDescriptor | None = None, thing: labthings_fastapi.thing.Thing | None = None) → list[labthings_fastapi.invocations.InvocationSummary]

All of the invocations currently managed.

Returns a list of InvocationSummary instances representing all the invocations that are currently running, or have recently completed and not yet expired.

Parameters:

action – filters out only the invocations of a particular ActionDescriptor. Note that if there are two Things of the same subclass, filtering by action will return invocations on either Thing.
thing – returns only invocations of actions on a particular Thing. This will often be combined with filtering by action to give the list of invocations returned by a GET request on an action endpoint.

Returns:

A list of invocations, optionally filtered by Thing and/or Action.

expire_invocations() → None: Delete invocations that have passed their expiry time.

router() → fastapi.APIRouter

Create a FastAPI Router with action-related endpoints.

Returns:: a Router with all action-related endpoints.