labthings_fastapi

LabThings-FastAPI.

This is the top level module for LabThings-FastAPI, a library for building Web of Things Core Concepts devices using Python. There is documentation on readthedocs, and the recommended place to start is labthings_fastapi.

This module contains a number of convenience imports and is intended to be imported using:

import labthings_fastapi as lt

The example code elsewhere in the documentation generally follows this convention. Symbols in the top-level module mostly exist elsewhere in the package, but should be imported from here as a preference, to ensure code does not break if modules are rearranged.

Submodules

• labthings_fastapi.actions
• labthings_fastapi.base_descriptor
• labthings_fastapi.client
• labthings_fastapi.dependencies
• labthings_fastapi.deps
• labthings_fastapi.endpoints
• labthings_fastapi.example_things
• labthings_fastapi.exceptions
• labthings_fastapi.invocation_contexts
• labthings_fastapi.invocations
• labthings_fastapi.logs
• labthings_fastapi.notifications
• labthings_fastapi.outputs
• labthings_fastapi.properties
• labthings_fastapi.server
• labthings_fastapi.testing
• labthings_fastapi.thing
• labthings_fastapi.thing_description
• labthings_fastapi.thing_server_interface
• labthings_fastapi.thing_slots
• labthings_fastapi.types
• labthings_fastapi.utilities
• labthings_fastapi.websockets

Classes

Thing	Represents a Thing, as defined by the Web of Things standard.
ThingServerInterface	An interface for Things to interact with their server.
DataProperty	A Property descriptor that acts like a regular variable.
DataSetting	A DataProperty that persists on disk.
ThingServer	Use FastAPI to serve Thing instances.
ThingConfig	The information needed to add a Thing to a ThingServer.
ThingServerConfig	The configuration parameters for a ThingServer.
ThingClient	A client for a LabThings-FastAPI Thing.
ThreadWithInvocationID	A thread that sets a new invocation ID.

Functions

thing_slot(→ Any)	Declare a connection to another Thing in the same server.
property(…)	Define a Property on a Thing.
setting(…)	Define a Setting on a Thing.
action(…)	Mark a method of a Thing as a LabThings Action.
endpoint(→ Callable[[Callable], EndpointDescriptor])	Mark a function as a FastAPI endpoint without making it an action.
cancellable_sleep(→ None)	Sleep for a specified time, allowing cancellation.
raise_if_cancelled(→ None)	Raise an exception if the current invocation has been cancelled.

Package Contents

class labthings_fastapi.Thing(thing_server_interface: labthings_fastapi.thing_server_interface.ThingServerInterface)

Represents a Thing, as defined by the Web of Things standard.

This class should encapsulate the code that runs a piece of hardware, or provides a particular function - it will correspond to a path on the server, and a Thing Description document.

Subclassing Notes

• __init__: You should accept any arguments you need to configure the Thing in __init__. Don’t initialise any hardware at this time, as your Thing may be instantiated quite early, or even at import time. You must make sure to call super().__init__(thing_server_interface).

• __enter__(self) and __exit__(self, exc_t, exc_v, exc_tb) are where you should start and stop communications with the hardware. This is Python’s “context manager” protocol. The arguments of __exit__ will be None except after errors. You should be safe to ignore them, and just include code that will close down your hardware, which is equivalent to a finally: block.

• Properties and Actions are defined using decorators: the @action decorator declares a method to be an action, which will run when it’s triggered, and the @property decorator does the same for a property.

  Properties may also be defined using dataclass-style syntax, if they do not need getter and setter functions.

  See the documentation on those functions for more detail.

• title will be used in various places as the human-readable name of your Thing, so it makes sense to set this in a subclass.

There are various LabThings methods that you should avoid overriding unless you know what you are doing: anything not mentioned above that’s defined in Thing is probably best left alone. They may in time be collected together into a single object to avoid namespace clashes.

Initialise a Thing.

The most important function of __init__ is attaching the thing_server_interface, and setting the path. Note that Thing instances are usually created by a ThingServer and not instantiated directly: if you do make a Thing directly, you will need to supply a ThingServerInterface that is connected to a ThingServer or a suitable mock object.

param thing_server_interface:

The interface to the server that is hosting this Thing. It will be supplied when the Thing is instantiated by the ThingServer or by create_thing_without_server which generates a mock interface.

title: str

A human-readable description of the Thing

_thing_server_interface: labthings_fastapi.thing_server_interface.ThingServerInterface

Provide access to features of the server that this Thing is attached to.

_disable_saving_settings: bool = False

property path: str

The path at which the Thing is exposed over HTTP.

property name: str

The name of this Thing, as known to the server.

property logger: logging.Logger

A logger, named after this Thing.

async __aenter__() → typing_extensions.Self

Context management is used to set up/close the thing.

As things (currently) do everything with threaded code, we define async __aenter__ and __aexit__ wrappers to call the synchronous code, if it exists.

Returns:

this object.

async __aexit__(exc_t: Any | None, exc_v: Any | None, exc_tb: Any) → None

Wrap context management functions, if they exist.

See __aenter__ for more details.

Parameters:

• exc_t – The type of the exception, or None.

• exc_v – The exception that occurred, or None.

• exc_tb – The traceback for the exception, or None.

attach_to_server(server: labthings_fastapi.server.ThingServer) → None

Attach this thing to the server.

Things need to be attached to a server before use to function correctly.

Parameters:

server – The server to attach this Thing to.

Attaching the Thing to a ThingServer allows the Thing to start actions, load its settings from the correct place, and create HTTP endpoints to allow it to be accessed from the HTTP API.

We create HTTP endpoints for all Interaction Affordances on the Thing, as well as any EndpointDescriptor descriptors.

_settings_store: dict[str, labthings_fastapi.properties.BaseSetting] | None = None

property _settings: dict[str, labthings_fastapi.properties.BaseSetting]

A private property that returns a dict of all settings for this Thing.

Each dict key is the name of the setting, the corresponding value is the BaseSetting class (a descriptor). This can be used to directly get the descriptor so that the value can be set without emitting signals, such as on startup.

load_settings() → None

Load settings from json.

Read the JSON file and use it to populate settings.

Note

Settings are loaded when the Thing is added to a server, so they will not be available while the __init__ method is run.

Note that no notifications will be triggered when the settings are set, so if action is needed (e.g. updating hardware with the loaded settings) it should be taken in __enter__.

save_settings() → None

Save settings to JSON.

This is called whenever a setting is updated. All settings are written to the settings file every time.

_labthings_thing_state: dict | None = None

property thing_state: collections.abc.Mapping

Return a dictionary summarising our current state.

This is intended to be an easy way to collect metadata from a Thing that summarises its state. It might be used, for example, to record metadata along with each reading/image/etc. when an instrument is saving data.

It’s best to populate this automatically so it can always be accessed. If it requires calls e.g. to a serial instrument, bear in mind it may be called quite often and shouldn’t take too long.

Some measure of caching here is a nice aim for the future, but not yet implemented.

validate_thing_description() → None

Raise an exception if the thing description is not valid.

_cached_thing_description: tuple[str | None, str | None, labthings_fastapi.thing_description._model.ThingDescription] | None = None

thing_description(path: str | None = None, base: str | None = None) → labthings_fastapi.thing_description._model.ThingDescription

Generate a w3c Thing Description representing this thing.

The w3c Web of Things working group defined a standard representation of a Thing, which provides a high-level description of the actions, properties, and events that it exposes. This endpoint delivers a JSON representation of the Thing Description for this Thing.

Parameters:

• path – the URL pointing to this Thing.

• base – the base URL for all URLs in the thing description.

Returns:

a Thing Description.

thing_description_dict(path: str | None = None, base: str | None = None) → dict

Describe this Thing with a Thing Description as a simple dict.

See Thing.thing_description. This function converts the return value of that function into a simple dictionary.

Parameters:

• path – the URL pointing to this Thing.

• base – the base URL for all URLs in the thing description.

Returns:

a Thing Description.

observe_property(property_name: str, stream: anyio.abc.ObjectSendStream) → None

Register a stream to receive property change notifications.

Parameters:

• property_name – the property to register for.

• stream – the stream used to send events.

Raises:

• KeyError – if the requested name is not defined on this Thing.

• PropertyNotObservableError – if the property is not observable.

observe_action(action_name: str, stream: anyio.abc.ObjectSendStream) → None

Register a stream to receive action status change notifications.

Parameters:

• action_name – the action to register for.

• stream – the stream used to send events.

Raises:

KeyError – if the requested name is not defined on this Thing.

labthings_fastapi.thing_slot(default: str | collections.abc.Iterable[str] | None | types.EllipsisType = ...) → Any

Declare a connection to another Thing in the same server.

lt.thing_slot marks a class attribute as a connection to another Thing on the same server. This will be automatically supplied when the server is started, based on the type hint and default value.

In keeping with property and setting, the type of the attribute should be the type of the connected Thing. For example:

import labthings_fastapi as lt


class ThingA(lt.Thing): ...


class ThingB(lt.Thing):
    "A class that relies on ThingA."

    thing_a: ThingA = lt.thing_slot()

This function is a convenience wrapper around the ThingSlot descriptor class, and should be used in preference to using the descriptor directly. The main reason to use the function is that it suppresses type errors when using static type checkers such as mypy or pyright (see note below).

The type hint of a Thing Connection should be one of the following:

• A Thing subclass. An instance of this subclass will be returned when

  the attribute is accessed.

• An optional Thing subclass (e.g. MyThing | None). This will either

  return a MyThing instance or None.

• A mapping of str to Thing (e.g. Mapping[str, MyThing]). This will

  return a mapping of Thing names to Thing instances. The mapping may be empty.

Example:

import labthings_fastapi as lt


class ThingA(lt.Thing):
    "An example Thing."


class ThingB(lt.Thing):
    "An example Thing with connections."

    thing_a: ThingA = lt.thing_slot()
    maybe_thing_a: ThingA | None = lt.thing_slot()
    all_things_a: Mapping[str, ThingA] = lt.thing_slot()

    @lt.action
    def show_connections(self) -> str:
        "Tell someone about our connections."
        self.thing_a  # should always evaluate to a ThingA instance
        self.maybe_thing_a  # will be a ThingA instance or None
        self.all_things_a  # will a mapping of names to ThingA instances
        return f"{self.thing_a=}, {self.maybe_thing_a=}, {self.all_things_a=}"

The example above is very contrived, but shows how to apply the different types.

If no default value is supplied, and no value is configured for the connection, the server will attempt to find a Thing that matches the specified type when the server is started. If no matching Thing instances are found, the descriptor will return None or an empty mapping. If that is not allowed by the type hint, the server will fail to start with an error.

The default value may be a string specifying a Thing name, or a sequence of strings (for connections that return mappings). In those cases, the relevant Thing will be returned from the server. If a name is given that either doesn’t correspond to a Thing on the server, or is a Thing that doesn’t match the type of this connection, the server will fail to start with an error.

The default may also be None which is appropriate when the type is optional or a mapping. If the type is a Thing subclass, a default value of None forces the connection to be specified in configuration.

Parameters:

default – The name(s) of the Thing(s) that will be connected by default. If the default is omitted or set to ... the server will attempt to find a matching Thing instance (or instances). A default value of None is allowed if the connection is type hinted as optional.

Returns:

A ThingSlot descriptor.

Typing notes:

In the example above, using ThingSlot directly would assign an object with type ThingSlot[ThingA] to the attribute thing_a, which is typed as ThingA. This would cause a type error. Using thing_slot suppresses this error, as its return type is a`Any``.

The use of Any or an alternative type-checking exemption seems to be inevitable when implementing descriptors that are typed via attribute annotations, and it is done by established libraries such as pydantic.

class labthings_fastapi.ThingServerInterface(server: labthings_fastapi.server.ThingServer, 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.

_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).

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.

get_thing_states() → Mapping[str, Any]

Retrieve metadata from all Things on the server.

This function will retrieve the Thing.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.

labthings_fastapi.property(getter: Callable[[Any], Value]) → FunctionalProperty[Value]

labthings_fastapi.property(*, default: Value, readonly: bool = False, **constraints: Any) → Value

labthings_fastapi.property(*, default_factory: Callable[[], Value], readonly: bool = False, **constraints: Any) → Value

Define a Property on a Thing.

This function may be used to define Properties in two ways, as either a decorator or a field specifier. See the examples in the property documentation.

Properties should always have a type annotation. This type annotation will be used in automatic documentation and also to serialise the value to JSON when it is sent over the network. This mean that the type of your property should either be JSON serialisable (i.e. simple built-in types) or a subclass of pydantic.BaseModel.

Parameters:

• getter – is a method of a class that returns the value of this property. This is usually supplied by using property as a decorator.

• default – is the default value. Either this, getter or default_factory must be specified. Specifying both or neither will raise an exception.

• default_factory – should return your default value. This may be used as an alternative to default if you need to use a mutable datatype. For example, it would be better to specify default_factory=list than default=[] because the second form would be shared between all Things with this property.

• readonly – whether the property should be read-only via the ThingClient interface (i.e. over HTTP or via a DirectThingClient). This is automatically true if property is used as a decorator and no setter is specified.

• **constraints – additional keyword arguments are passed to pydantic.Field and allow constraints to be added to the property. For example, ge=0 constrains a numeric property to be non-negative. See pydantic.Field for the full range of constraint arguments.

Returns:

a property descriptor, either a FunctionalProperty if used as a decorator, or a DataProperty if used as a field.

Raises:

• MissingDefaultError – if no valid default value is supplied, and a getter is not in use.

• OverspecifiedDefaultError – if the default is specified more than once (e.g. default, default_factory, or getter).

Typing Notes

This function has somewhat complicated type hints, for two reasons. Firstly, it may be used either as a decorator or as a field specifier, so default performs double duty as a default value or a getter. Secondly, when used as a field specifier the type hint for the property is attached to the attribute of the class to which the function’s output is assigned. This means property does not know its type hint until after it’s been called.

When used as a field specifier, property returns a generic DataProperty descriptor instance, which will determine its type when it is attached to the Thing. The type hint on the return value of property in that situation is a “white lie”: we annotate the return as having the same type as the default value (or the default_factory return value). This means that type checkers such as mypy will check that the default is valid for the type of the field, and won’t raise an error about assigning, for example, an instance of DataProperty[int] to a field annotated as int.

Finally, the type of the default argument includes EllipsisType so that we can use ... as its default value. This allows us to distinguish between default not being set (...) and a desired default value of None. Similarly, ... is the default value for getter so we can raise a more helpful error if a non-callable value is passed as the first argument.

labthings_fastapi.setting(getter: Callable[[Any], Value]) → FunctionalSetting[Value]

labthings_fastapi.setting(*, default: Value, readonly: bool = False, **constraints: Any) → Value

labthings_fastapi.setting(*, default_factory: Callable[[], Value], readonly: bool = False, **constraints: Any) → Value

Define a Setting on a Thing.

A setting is a property that is saved to disk.

This function defines a setting, which is a special Property that will be saved to disk, so it persists even when the LabThings server is restarted. It is otherwise very similar to property.

A type annotation is required, and should follow the same constraints as for @property.

Every setting on a Thing will be read each time the settings are saved, which may be quite frequent. This means your getter must not take too long to run, or have side-effects. Settings that use getters and setters may be removed in the future pending the outcome of #159.

If the type is a pydantic BaseModel, then the setter must also be able to accept the dictionary representation of this BaseModel as this is what will be used to set the Setting when loading from disk on starting the server.

Note

If a setting is mutated rather than set, this will not trigger saving. For example: if a Thing has a setting called dictsetting holding the dictionary {"a": 1, "b": 2} then self.dictsetting = {"a": 2, "b": 2} would trigger saving but self.dictsetting[a] = 2 would not, as the setter for dictsetting is never called.

Parameters:

• getter – is a method of a class that returns the value of this property. This is usually supplied by using property as a decorator.

• default – is the default value. Either this, getter or default_factory must be specified. Specifying both or neither will raise an exception.

• default_factory – should return your default value. This may be used as an alternative to default if you need to use a mutable datatype. For example, it would be better to specify default_factory=list than default=[] because the second form would be shared between all Things with this setting.

• readonly – whether the setting should be read-only via the ThingClient interface (i.e. over HTTP or via a DirectThingClient).

• **constraints – additional keyword arguments are passed to pydantic.Field and allow constraints to be added to the property. For example, ge=0 constrains a numeric property to be non-negative. See pydantic.Field for the full range of constraint arguments.

Returns:

a setting descriptor.

Raises:

• MissingDefaultError – if no valid default or getter is supplied.

• OverspecifiedDefaultError – if the default is specified more than once (e.g. default, default_factory, or getter).

Typing Notes

See the typing notes on property as they all apply to setting as well.

class labthings_fastapi.DataProperty(default: Value, *, readonly: bool = False, constraints: collections.abc.Mapping[str, Any] | None = None)

class labthings_fastapi.DataProperty(*, default_factory: ValueFactory, readonly: bool = False, constraints: collections.abc.Mapping[str, Any] | None = None)

Bases: BaseProperty[Value], Generic[Value]

A Property descriptor that acts like a regular variable.

DataProperty descriptors remember their value, and can be read and written to like a regular Python variable.

Create a property that acts like a regular variable.

DataProperty descriptors function just like variables, in that they can be read and written to as attributes of the Thing and their value will be the same every time it is read (i.e. it changes only when it is set). This differs from FunctionalProperty which uses a “getter” function just like builtins.property and may return a different value each time.

DataProperty instances may always be set, when they are accessed as an attribute of the Thing instance. The readonly parameter applies only to client code, whether it is remote or a DirectThingClient wrapper.

The type of the property’s value will be inferred either from the type subscript or from an annotation on the class attribute. This is done in __get_name__ because neither is available during __init__.

Parameters:

• default – the default value. This or default_factory must be provided. Note that, as None is a valid default value, this uses ... instead as a way of checking whether default has been set.

• default_factory – a function that returns the default value. This is appropriate for datatypes such as lists, where using a mutable default value can lead to odd behaviour.

• readonly – if True, the property may not be written to via HTTP, or via DirectThingClient objects, i.e. it may only be set as an attribute of the Thing and not from a client.

• constraints – is passed as keyword arguments to pydantic.Field to add validation constraints to the property. See pydantic.Field for details.

_default_factory = None

readonly = False

instance_get(obj: labthings_fastapi.thing.Thing) → Value

Return the property’s value.

This will supply a default if the property has not yet been set.

Parameters:

obj – The Thing on which the property is being accessed.

Returns:

the value of the property.

__set__(obj: labthings_fastapi.thing.Thing, value: Value, emit_changed_event: bool = True) → None

Set the property’s value.

This sets the property’s value, and notifies any observers.

Parameters:

• obj – the Thing to which we are attached.

• value – the new value for the property.

• emit_changed_event – whether to emit a changed event.

_observers_set(obj: labthings_fastapi.thing.Thing) → weakref.WeakSet

Return the observers of this property.

Each observer in this set will be notified when the property is changed. See .DataProperty.emit_changed_event

Parameters:

obj – the Thing to which we are attached.

Returns:

the set of observers corresponding to obj.

emit_changed_event(obj: labthings_fastapi.thing.Thing, value: Value) → None

Notify subscribers that the property has changed.

This function is run when properties are updated. It must be run from within a thread. This could be the Invocation thread of a running action, or the property should be updated over via a client/http. It must be run from a thread as it is communicating with the event loop via an asyncio blocking portal and can cause deadlock if run in the event loop.

This method will raise a ServerNotRunningError if the event loop is not running, and should only be called after the server has started.

Parameters:

• obj – the Thing to which we are attached.

• value – the new property value, to be sent to observers.

async emit_changed_event_async(obj: labthings_fastapi.thing.Thing, value: Value) → None

Notify subscribers that the property has changed.

This function may only be run in the anyio event loop. See DataProperty.emit_changed_event.

Parameters:

• obj – the Thing to which we are attached.

• value – the new property value, to be sent to observers.

class labthings_fastapi.DataSetting(default: Value, *, readonly: bool = False, constraints: collections.abc.Mapping[str, Any] | None = None)

class labthings_fastapi.DataSetting(*, default_factory: ValueFactory, readonly: bool = False, constraints: collections.abc.Mapping[str, Any] | None = None)

Bases: DataProperty[Value], BaseSetting[Value], Generic[Value]

A DataProperty that persists on disk.

A setting can be accessed via the HTTP API and is persistent between sessions.

A DataSetting is a DataProperty with extra functionality for triggering a Thing to save its settings.

Note: If a setting is mutated rather than assigned to, this will not trigger saving. For example: if a Thing has a setting called dictsetting holding the dictionary {"a": 1, "b": 2} then self.dictsetting = {"a": 2, "b": 2} would trigger saving but self.dictsetting[a] = 2 would not, as the setter for dictsetting is never called.

The setting otherwise acts just like a normal variable.

Create a property that acts like a regular variable.

DataProperty descriptors function just like variables, in that they can be read and written to as attributes of the Thing and their value will be the same every time it is read (i.e. it changes only when it is set). This differs from FunctionalProperty which uses a “getter” function just like builtins.property and may return a different value each time.

DataProperty instances may always be set, when they are accessed as an attribute of the Thing instance. The readonly parameter applies only to client code, whether it is remote or a DirectThingClient wrapper.

The type of the property’s value will be inferred either from the type subscript or from an annotation on the class attribute. This is done in __get_name__ because neither is available during __init__.

Parameters:

• default – the default value. This or default_factory must be provided. Note that, as None is a valid default value, this uses ... instead as a way of checking whether default has been set.

• default_factory – a function that returns the default value. This is appropriate for datatypes such as lists, where using a mutable default value can lead to odd behaviour.

• readonly – if True, the property may not be written to via HTTP, or via DirectThingClient objects, i.e. it may only be set as an attribute of the Thing and not from a client.

• constraints – is passed as keyword arguments to pydantic.Field to add validation constraints to the property. See pydantic.Field for details.

__set__(obj: labthings_fastapi.thing.Thing, value: Value, emit_changed_event: bool = True) → None

Set the setting’s value.

This will cause the settings to be saved to disk.

Parameters:

• obj – the Thing to which we are attached.

• value – the new value of the setting.

• emit_changed_event – whether to emit a changed event.

set_without_emit(obj: labthings_fastapi.thing.Thing, value: Value) → None

Set the property’s value, but do not emit event to notify the server.

This function is not expected to be used externally. It is called during initial setup so that the setting can be set from disk before the server is fully started.

Parameters:

• obj – the Thing to which we are attached.

• value – the new value of the setting.

labthings_fastapi.action(func: Callable[Concatenate[OwnerT, ActionParams], ActionReturn], **kwargs: Any) → ActionDescriptor[ActionParams, ActionReturn, OwnerT]

labthings_fastapi.action(**kwargs: Any) → Callable[[Callable[Concatenate[OwnerT, ActionParams], ActionReturn]], ActionDescriptor[ActionParams, ActionReturn, OwnerT]]

Mark a method of a Thing as a LabThings Action.

Methods decorated with @action will be available to call over HTTP as actions. See Actions for an introduction to the concept of actions.

This decorator may be used with or without arguments.

Parameters:

• func – The method to be decorated as an action.

• **kwargs – Keyword arguments are passed to the constructor of ActionDescriptor.

Returns:

Whether used with or without arguments, the result is that the method is wrapped in an ActionDescriptor, so it can be called as usual, but will also be exposed over HTTP.

labthings_fastapi.endpoint(method: HTTPMethod, path: str | None = None, **kwargs: Any) → Callable[[Callable], EndpointDescriptor]

Mark a function as a FastAPI endpoint without making it an action.

This decorator will cause a method of a Thing to be directly added to the HTTP API, bypassing the machinery underlying Action and Property affordances. Such endpoints will not be documented in the Thing Description but may be used as the target of links. For example, this could allow a file to be downloaded from the Thing at a known URL, or serve a video stream that wouldn’t be supported as a Blob.

The majority of Thing implementations won’t need this decorator, but it is here to enable flexibility when it’s needed.

This decorator always takes arguments; in particular, method is required. It should be used as:

class DownloadThing(Thing):
    @endpoint("get")
    def plain_text_response(self) -> str:
        return "example string"

This decorator is intended to work very similarly to the fastapi decorators @app.get, @app.post, etc., with two changes:

1. The path is relative to the host Thing and will default to the name

   of the method.

2. The method will be called with the host Thing as its first argument,

   i.e. it will be bound to the class as usua.

Parameters:

• method – The HTTP verb this endpoint responds to.

• path – The path, relative to the host Thing base URL.

• **kwargs – Additional keyword arguments are passed to the fastapi.FastAPI.get decorator if method is get, or to the equivalent decorator for other HTTP verbs.

Returns:

When used as intended, the result is an EndpointDescriptor.

class labthings_fastapi.ThingServer(things: config_model.ThingsConfig, settings_folder: str | None = None)

Use FastAPI to serve Thing instances.

The ThingServer sets up a fastapi.FastAPI application and uses it to expose the capabilities of Thing instances over HTTP.

There are several functions of a ThingServer:

• Manage where settings are stored, to allow Thing instances to load and save their settings from disk.

• Configure the server to allow cross-origin requests (required if we use a web app that is not served from the ThingServer).

• Manage the threads used to run Actions.

• Manage Blob input/output to allow binary data to be returned.

• Allow threaded code to call functions in the event loop, by providing an anyio.from_thread.BlockingPortal.

Initialise a LabThings server.

Setting up the ThingServer involves creating the underlying fastapi.FastAPI app, setting its lifespan function (used to set up and shut down the Thing instances), and configuring it to allow cross-origin requests.

We also create the ActionManager to manage Actions and the BlobManager to manage the downloading of Blob input/output.

Parameters:

• things – A mapping of Thing names to Thing subclasses, or ThingConfig objects specifying the subclass, its initialisation arguments, and any connections to other Things.

• settings_folder – the location on disk where Thing settings will be saved.

_config

app

settings_folder = './settings'

action_manager

blocking_portal: anyio.from_thread.BlockingPortal | None = None

startup_status: dict[str, str | dict]

_things

classmethod from_config(config: config_model.ThingServerConfig) → typing_extensions.Self

Create a ThingServer from a configuration model.

This is equivalent to ThingServer(**dict(config)).

Parameters:

config – The configuration parameters for the server.

Returns:

A ThingServer configured as per the model.

_set_cors_middleware() → None

Configure the server to allow requests from other origins.

This is required to allow web applications access to the HTTP API, if they are not served from the same origin (i.e. if they are not served as part of the ThingServer.).

This is usually needed during development, and may be needed at other times depending on how you are using LabThings.

property things: collections.abc.Mapping[str, labthings_fastapi.thing.Thing]

Return a dictionary of all the things.

Returns:

a dictionary mapping thing paths to Thing instances.

ThingInstance

things_by_class(cls: type[ThingInstance]) → collections.abc.Sequence[ThingInstance]

Return all Things attached to this server matching a class.

Return all instances of cls attached to this server.

Parameters:

cls – A Thing subclass.

Returns:

all instances of cls that have been added to this server.

thing_by_class(cls: type[ThingInstance]) → ThingInstance

Return the instance of cls attached to this server.

This function calls ThingServer.things_by_class, but asserts that there is exactly one match.

Parameters:

cls – a Thing subclass.

Returns:

the instance of cls attached to this server.

Raises:

RuntimeError – if there is not exactly one matching Thing.

path_for_thing(name: str) → str

Return the path for a thing with the given name.

Parameters:

name – The name of the thing.

Returns:

The path at which the thing is served.

Raises:

KeyError – if no thing with the given name has been added.

_create_things() → collections.abc.Mapping[str, labthings_fastapi.thing.Thing]

Create the Things, add them to the server, and connect them up if needed.

This method is responsible for creating instances of Thing subclasses and adding them to the server. It also ensures the Things are connected together if required.

The Things are defined in self._config.thing_configs which in turn is generated from the things argument to __init__.

Returns:

A mapping of names to Thing instances.

Raises:

TypeError – if cls is not a subclass of Thing.

_connect_things() → None

Connect the thing_slot attributes of Things.

A Thing may have attributes defined as lt.thing_slot(), which will be populated after all Thing instances are loaded on the server.

This function is responsible for supplying the Thing instances required for each connection. This will be done by using the name specified either in the connection’s default, or in the configuration of the server.

ThingSlotError will be raised by code called by this method if the connection cannot be provided. See ThingSlot.connect for more details.

_attach_things_to_server() → None

Add the Things to the FastAPI App.

This calls Thing.attach_to_server on each Thing that is a part of this ThingServer in order to add the HTTP endpoints and load settings.

async lifespan(app: fastapi.FastAPI) → AsyncGenerator[None, None]

Manage set up and tear down of the server and Things.

This method is used as a lifespan function for the FastAPI app. See the lifespan page in FastAPI’s documentation.

This does two important things:

• It sets up the blocking portal so background threads can run async code (this is required for events, streams, etc.).

• It runs setup/teardown code for Things by calling them as context managers.

Parameters:

app – The FastAPI application wrapped by the server.

Yield:

no value. The FastAPI application will serve requests while this function yields.

_add_things_view_to_app() → None

Add an endpoint that shows the list of attached things.

class labthings_fastapi.ThingConfig(/, **data: Any)

Bases: pydantic.BaseModel

The information needed to add a Thing to a ThingServer.

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.

cls: pydantic.ImportString = None

args: collections.abc.Sequence[Any] = None

kwargs: collections.abc.Mapping[str, Any] = None

thing_slots: collections.abc.Mapping[str, str | collections.abc.Iterable[str] | None] = None

class labthings_fastapi.ThingServerConfig(/, **data: Any)

Bases: pydantic.BaseModel

The configuration parameters for a ThingServer.

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.

things: ThingsConfig = None

classmethod check_things(things: ThingsConfig) → ThingsConfig

Check that the thing configurations can be normalised.

It’s possible to specify the things as a mapping from names to classes. We use pydantic.ImportString as the type of the classes: this takes a string, and imports the corresponding Python object. When loading config from JSON, this does the right thing - but when loading from Python objects it will accept any Python object.

This validator runs normalise_thing_config to check each value is either a valid ThingConfig or a type or a mapping. If it’s a mapping, we will attempt to make a ThingConfig from it. If it’s a type we will create a ThingConfig using that type as the class. We don’t check for Thing subclasses in this module to avoid a dependency loop.

Parameters:

things – The validated value of the field.

Returns:

A copy of the input, with all values converted to ThingConfig instances.

property thing_configs: collections.abc.Mapping[ThingName, ThingConfig]

A copy of the things field where every value is a .ThingConfig.

The field validator on things already ensures it returns a mapping, but it’s not typed strictly, to allow Things to be specified with just a class.

This property returns the list of ThingConfig objects, and is typed strictly.

settings_folder: str | None = None

class labthings_fastapi.ThingClient(base_url: str, client: httpx.Client | None = None)

A client for a LabThings-FastAPI Thing.

Note

ThingClient must be subclassed to add actions/properties, so this class will be minimally useful on its own.

The best way to get a client for a particular Thing is currently ThingClient.from_url, which dynamically creates a subclass with the right attributes.

Create a ThingClient connected to a remote Thing.

Parameters:

• base_url – the base URL of the Thing. This should be the URL of the Thing Description document.

• client – an optional httpx.Client object to use for all HTTP requests. This may be a fastapi.TestClient object for testing purposes.

server

path

client

get_property(path: str) → Any

Make a GET request to retrieve the value of a property.

Parameters:

path – the URI of the getproperty endpoint, relative to the base_url.

Returns:

the property’s value, as deserialised from JSON.

set_property(path: str, value: Any) → None

Make a PUT request to set the value of a property.

Parameters:

• path – the URI of the getproperty endpoint, relative to the base_url.

• value – the property’s value. Currently this must be serialisable to JSON.

invoke_action(path: str, **kwargs: Any) → Any

Invoke an action on the Thing.

This method will make the initial POST request to invoke an action, then poll the resulting invocation until it completes. If successful, the action’s output will be returned directly.

While the action is running, log messages will be re-logged locally. If you have enabled logging to the console, these should be visible.

Parameters:

• path – the URI of the invokeaction endpoint, relative to the base_url

• **kwargs – Additional arguments will be combined into the JSON body of the POST request and sent as input to the action. These will be validated on the server.

Returns:

the output value of the action.

Raises:

RuntimeError – is raised if the action does not complete successfully.

follow_link(response: dict, rel: str) → httpx.Response

Follow a link in a response object, by its rel attribute.

Parameters:

• response – is the dictionary returned by e.g. poll_invocation.

• rel – picks the link to follow by matching its rel item.

Returns:

the response to making a GET request to the link.

classmethod from_url(thing_url: str, client: httpx.Client | None = None) → typing_extensions.Self

Create a ThingClient from a URL.

This will dynamically create a subclass with properties and actions, and return an instance of that subclass pointing at the Thing URL.

Parameters:

• thing_url – The base URL of the Thing, which should also be the URL of its Thing Description.

• client – is an optional httpx.Client object. If not present, one will be created. This is particularly useful if you need to set HTTP options, or if you want to work with a local server object for testing purposes (see fastapi.TestClient).

Returns:

a ThingClient subclass with properties and methods that match the retrieved Thing Description (see Thing).

classmethod subclass_from_td(thing_description: dict) → type[typing_extensions.Self]

Create a ThingClient subclass from a Thing Description.

Dynamically subclass ThingClient to add properties and methods for each property and action in the Thing Description.

Parameters:

thing_description – A Thing Description as a dictionary, which will be used to construct the class.

Returns:

a ThingClient subclass with the right properties and methods.

labthings_fastapi.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.

Parameters:

interval – The length of time to wait for, in seconds.

labthings_fastapi.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 cancellable_sleep but without waiting any time.

If called outside of an invocation context, this function does nothing, and will not raise an error.

class labthings_fastapi.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: 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.

Parameters:

• target – the function to call in the thread.

• args – positional arguments to target.

• kwargs – keyword arguments to target.

• *super_args – arguments passed to threading.Thread.

• **super_kwargs – keyword arguments passed to threading.Thread.

_target

_args = []

_kwargs

_invocation_id: uuid.UUID

_result: Any = None

_exception: BaseException | None = None

_cancel_event

property invocation_id: uuid.UUID

The InvocationID of this thread.

property result: Any

The return value of the target function.

property exception: BaseException | None

The exception raised by the target function, or None.

cancel() → None

Set the cancel event to tell the code to terminate.

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.

Parameters:

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.

run() → None

Run the target function, with invocation ID set in the context variable.