labthings_fastapi.actions

Actions module.

Actions are represented by methods, decorated with the thing_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 thing_action and the underlying ActionDescriptor. This is likely to be refactored in the near future.

Submodules

labthings_fastapi.actions.invocation_model

Attributes

`blobdata_to_url_ctx`	This context variable gives access to a function that makes BlobData objects
`ACTION_INVOCATIONS_PATH`	The API route used to list `Invocation` objects.

Exceptions

`InvocationCancelledError`	An invocation was cancelled by the user.
`NoBlobManagerError`	Raised if an API route accesses Invocation outputs without a BlobIOContextDep.

Classes

`EmptyInput`	Represent the input of an action that has no required parameters.
`LinkElement`	See https://www.w3.org/TR/wot-thing-description11/#link.
`InvocationStatus`	The current status of an `Invocation`.
`LogRecordModel`	A model to serialise `logging.LogRecord` objects.
`Invocation`	A Thread subclass that retains output values and tracks progress.
`DequeLogHandler`	A log handler that stores entries in memory.
`ActionManager`	A class to manage a collection of actions.

Functions

model_to_dict(→ Dict[str, Any])

Convert a pydantic model to a dictionary, non-recursively.

Package Contents

labthings_fastapi.actions.model_to_dict(model: pydantic.BaseModel | None) → Dict[str, Any]

Convert a pydantic model to a dictionary, non-recursively.

We convert only the top level model, i.e. we do not recurse into submodels. This is important to avoid serialising Blob objects in action inputs. This function returns dict(model), with exceptions for the case of None (converted to an empty dictionary) and pydantic.RootModel (checked to see if they correspond to empty input).

If pydantic.RootModel with non-empty input is allowed, this function will need to be updated to handle them.

Parameters:: model – A Pydantic model (usually the input of an action).
Returns:: A dictionary with string keys, which are the fields of the model. This should be suitable for using as **kwargs to an action.
Raises:: ValueError – if we are given a root model that isn’t empty.

class labthings_fastapi.actions.EmptyInput(/, root=PydanticUndefined, **data)

Bases: pydantic.RootModel

Represent the input of an action that has no required parameters.

This may be either a dictionary or None.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

root: EmptyObject | None = None

class labthings_fastapi.actions.LinkElement(/, **data: Any)

Bases: pydantic.BaseModel

See https://www.w3.org/TR/wot-thing-description11/#link.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

href: AnyUri

type: str | None = None

rel: str | None = None

anchor: AnyUri | None = None

class labthings_fastapi.actions.InvocationStatus(*args, **kwds)

Bases: enum.Enum

The current status of an Invocation.

PENDING = 'pending': The Invocation has not yet been started.

RUNNING = 'running': The Invocation is running in its thread.

COMPLETED = 'completed': The Invocation finished successfully. A return value may be available.

CANCELLED = 'cancelled': The Invocation was cancelled and has finished.

ERROR = 'error': The Invocation terminated unexpectedly due to an error.

class labthings_fastapi.actions.LogRecordModel(/, **data: Any)

Bases: pydantic.BaseModel

A model to serialise logging.LogRecord objects.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

message: str

levelname: str

levelno: int

lineno: int

filename: str

created: datetime.datetime

classmethod generate_message(data: Any) → Any

Ensure LogRecord objects have constructed their message.

Parameters:: data – The LogRecord to process.
Returns:: The LogRecord, with a message constructed.

exception labthings_fastapi.actions.InvocationCancelledError

Bases: BaseException

An invocation was cancelled by the user.

Note that this inherits from BaseException so won’t be caught by except Exception, it must be handled specifically.

Action code may want to handle cancellation gracefully. This exception should be propagated if the action’s status should be reported as cancelled, or it may be handled so that the action finishes, returns a value, and is marked as completed.

If this exception is handled, the CancelEvent should be reset to allow another InvocationCancelledError to be raised if the invocation receives a second cancellation signal.

Initialize self. See help(type(self)) for accurate signature.

labthings_fastapi.actions.blobdata_to_url_ctx

This context variable gives access to a function that makes BlobData objects downloadable, by assigning a URL and adding them to the [BlobDataManager](#labthings_fastapi.outputs.blob.BlobDataManager).

It is only available within a [blob_serialisation_context_manager](#labthings_fastapi.outputs.blob.blob_serialisation_context_manager) because it requires access to the BlobDataManager and the url_for function from the FastAPI app.

labthings_fastapi.actions.ACTION_INVOCATIONS_PATH = '/action_invocations': The API route used to list Invocation objects.

exception labthings_fastapi.actions.NoBlobManagerError

Bases: RuntimeError

Raised if an API route accesses Invocation outputs without a BlobIOContextDep.

Any access to an invocation output must have BlobIOContextDep as a dependency, as the output may be a blob, and the blob needs this context to resolve its URL.

Initialize self. See help(type(self)) for accurate signature.

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

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.
cancel_hook – is a threading.Event subclass that tells the invocation it’s time to stop. See CancelHook.

action_ref

thing_ref

input = None

dependencies = None

cancel_hook = None

_ID

retention_time

expiry_time: datetime.datetime | None = None

_status_lock

_status: invocation_model.InvocationStatus

_return_value: Any | 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.

Raises:: NoBlobManagerError – If this is called in a context where the blob manager context variables are not available. This stops errors being raised later once the blob is returned and tries to serialise. If the errors happen during serialisation the stack-trace will not clearly identify the route with the missing dependency.

property log: list[invocation_model.LogRecordModel]: A list of log items generated by the Action.

property status: invocation_model.InvocationStatus

Current running status of the thread.

See InvocationStatus for the values and their meanings.

property action: The ActionDescriptor object running in this thread.

property thing: labthings_fastapi.thing.Thing: The Thing to which the action is bound, i.e. this is self.

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.

response(request: fastapi.Request | None = None) → invocation_model.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.

Parameters:: request – is used to generate the href in the response, which should retrieve an updated version of this response.
Returns:: an InvocationModel representing this Invocation.

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:: Exception – any exception raised in the action function will propagate through this method. Usually, this will just cause the thread to terminate after setting status to ERROR and saving the exception to self._exception.

class labthings_fastapi.actions.DequeLogHandler(dest: MutableSequence, level: int = logging.INFO)

Bases: logging.Handler

A log handler that stores entries in memory.

Set up a log handler that appends messages to a deque.

Warning

This log handler does not currently rotate or truncate the list - so if you use it on a thread that produces a lot of log messages, you may run into memory problems.

Using a deque with a finite capacity helps to mitigate this.

Parameters:

dest – should specify a deque, to which we will append each log entry as it comes in. This is assumed to be thread safe.
level – sets the level of the logger. For most invocations, a log level of logging.INFO is appropriate.

dest

emit(record: logging.LogRecord) → None

Save a log record to the destination deque.

Parameters:: record – the logging.LogRecord object to add.

class labthings_fastapi.actions.ActionManager

A class to manage a collection of actions.

Set up an ActionManager.

_invocations

_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: labthings_fastapi.descriptors.ActionDescriptor, thing: labthings_fastapi.thing.Thing, id: uuid.UUID, input: Any, dependencies: dict[str, Any], cancel_hook: labthings_fastapi.dependencies.invocation.CancelHook) → 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.
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.
cancel_hook – is a threading.Event subclass that tells the invocation it’s time to stop. See CancelHook.

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: labthings_fastapi.descriptors.ActionDescriptor | None = None, thing: labthings_fastapi.thing.Thing | None = None, request: fastapi.Request | None = None) → list[invocation_model.InvocationModel]

All of the invocations currently managed.

Returns a list of InvocationModel 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.
request – is used to pass a fastapi.Request object to the Invocation.response method. Doing so ensures the URL returned as href in the response matches the address used to communicate with the server (i.e. it uses fastapi.Request.url_for instead of a path generated from a string).

Returns:

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

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

attach_to_app(app: fastapi.FastAPI) → None

Add /action_invocations and /action_invocation/{id} endpoints to FastAPI.

Parameters:: app – The fastapi.FastAPI application to which we add the endpoints.