labthings_fastapi.thing_server_interface

Interface between Thing subclasses and the ThingServer.

Attributes

`Params`
`ReturnType`

Exceptions

ThingServerMissingError

The error raised when a ThingServer is no longer available.

Classes

ThingServerInterface

An interface for Things to interact with their server.

Module Contents

labthings_fastapi.thing_server_interface.Params

labthings_fastapi.thing_server_interface.ReturnType

exception labthings_fastapi.thing_server_interface.ThingServerMissingError

Bases: RuntimeError

The error raised when a ThingServer is no longer available.

This error indicates that a ThingServerInterface is still in use even though its underlying ThingServer has been deleted. This is unlikely to happen and usually indicates that the server has been created in an odd way.

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

class labthings_fastapi.thing_server_interface.ThingServerInterface(server: labthings_fastapi.server.ThingServer, name: str, class_name: str)

An interface for Things to interact with their server.

This is added to every Thing during __init__ and is available as self._thing_server_interface.

Initialise a ThingServerInterface.

The ThingServerInterface sits between a Thing and its ThingServer, with the intention of providing a useful set of functions, without exposing too much of the server to the Thing.

One reason for using this intermediary class is to make it easier to mock the server during testing: only functions provided here need be mocked, not the whole functionality of the server.

Parameters:

server – the ThingServer instance we’re connected to. This will be retained as a weak reference.
name – the name of the Thing instance this interface is provided for.
class_name – The name of the class of the Thing, used as part of the settings filename.

_name: str

_class_name: str

_server: weakref.ReferenceType[labthings_fastapi.server.ThingServer]

_get_server() → labthings_fastapi.server.ThingServer

Return a live reference to the ThingServer.

This will evaluate the weak reference to the ThingServer, and will raise an exception if the server has been garbage collected.

The server is, in practice, not going to be finalized before the Things, so this should not be a problem.

Returns:: the ThingServer.
Raises:: ThingServerMissingError – if the ThingServer is no longer available.

start_async_task_soon(async_function: Callable[Params, Awaitable[ReturnType]], *args: Any) → concurrent.futures.Future[ReturnType]

Run an asynchronous task in the server’s event loop.

This function wraps anyio.from_thread.BlockingPortal.start_task_soon to provide a way of calling asynchronous code from threaded code. It will call the provided async function in the server’s event loop, without any guarantee of exactly when it will happen. This means we will return immediately, and the return value of this function will be a concurrent.futures.Future object that may resolve to the async function’s return value.

Parameters:

async_function – the asynchronous function to call.
*args – positional arguments to be provided to the function.

Returns:

an asyncio.Future object wrapping the return value.

Raises:

ServerNotRunningError – if the server is not running (i.e. there is no event loop).

call_async_task(async_function: Callable[Params, Awaitable[ReturnType]], *args: Any) → ReturnType

Run an asynchronous task in the server’s event loop in a blocking manner.

This function wraps anyio.from_thread.BlockingPortal.call to provide a way of calling asynchronous code from threaded code. It will block the current thread while it calls the provided async function in the server’s event loop.

Do not call this from the event loop or it may lead to a deadlock.

Parameters:

async_function – the asynchronous function to call.
*args – positional arguments to be provided to the function.

Returns:

The return value from the asynchronous function.

Raises:

ServerNotRunningError – if the server is not running (i.e. there is no event loop).

publish(message: labthings_fastapi.message_broker.Message) → None

Publish an event.

Use the async event loop to notify subscribers that something has happened. The message should contain the name of the Thing and affordance.

Note that this function will do nothing if the event loop is not yet running.

Parameters:: message – the message being published.

property settings_folder: str: The path to a folder where persistent files may be saved.

property settings_file_path: str: The path where settings should be loaded and saved as JSON.

property name: str: The name of the Thing attached to this interface.

property path: str

The path, relative to the server’s base URL, of the Thing.

A ThingServerInterface is specific to one Thing, so this path points to the base URL of the Thing, i.e. the Thing Description’s endpoint.

property application_config: Mapping[str, Any] | None: The custom application configuration options from configuration.

get_thing_states() → Mapping[str, Any]

Retrieve metadata from all Things on the server.

This function will retrieve the thing_state property from each Thing on the server, and return it as a dictionary. It is intended to make it easy to add metadata to the results of actions, for example to embed in an image.

Returns:: a dictionary of metadata, with the Thing names as keys.

property _action_manager: labthings_fastapi.actions.ActionManager

The ActionManager for the Thing attached to this interface.

This property may be removed in future, and is for internal use only.

property global_lock: labthings_fastapi.global_lock.GlobalLock | None

A lock that ensures property writes and actions are one-at-a-time.

If global locking is not enabled, this property will return None.

_optionally_hold_global_lock(enabled: bool | None = True) → collections.abc.Iterator[None]

Hold the global lock, if required, as a context manager.

This function will hold the global lock if necessary while a block of code runs. Its behaviour is controlled by the enabled parameter: if enabled is False this function does nothing. If it is None (the default when called from a property or action that’s not otherwise configured), the global lock is held if it exists, but no error is raised if global locking is disabled.

If enabled is True (the default if no arguments are passed), an error will be raised if there is no global lock.

Parameters:: enabled – whether to use the global lock. True and False have the obvious meanings described above, None will use the lock if it is enabled globally but won’t raise an error if it is unavailable.
Raises:: FeatureNotEnabledError – if enabled is True but the global lock is not enabled.

hold_global_lock(*, error_if_unavailable: bool = True) → collections.abc.Iterator[None]

Hold the global lock for the duration of a with block.

This context manager will hold the global lock while a with: block runs. By default, an exception will be raised if the global lock is not enabled.

Parameters:: error_if_unavailable – may be set to False to suppress errors if the global lock is not enabled. This means the context manager silently does nothing, if the global lock is not available.