Public API Documentation

This page summarises the parts of the LabThings API that should be most frequently used by people writing lt.Thing subclasses. It doesn’t list options exhaustively: the full API documentation does that if extra detail is needed.

class lt.Thing(thing_server_interface: ThingServerInterface)

The basic unit of functionality in LabThings is the Thing. Each piece of hardware (or software) controlled by LabThings is represented by an instance of a Thing subclass, and so adding a new instrument or software unit generally involves subclassing. Each Thing on a server has a name, a URL, and a Thing Description describing its capabilities.

It is likely that Thing subclasses will override __init__, __enter__, and __exit__. See the documentation on those methods for important subclassing notes.

The capabilities of a Thing are described using attributes. Actions are methods decorated with action, and properties are declared using property or setting. Things may communicate with each other using thing_slots.

Things should only be created by a ThingServer. To create a Thing without a server, see testing.create_thing_without_server for a test harness that supplies a dummy ThingServerInterface.

This page offers only the most commonly-used methods: full documentation is available at labthings_fastapi.thing.Thing.

__init__(thing_server_interface: ThingServerInterface)

__init__ may be overridden in order to accept arguments when your object is created. If you override __init__ you must call super().__init__(thing_server_interface) to ensure the Thing is properly connected to a server.

__init__ should not acquire any resources (like communications ports), as they may not be closed cleanly. Please acquire any necessary resources in __enter__ instead.

Parameters:: 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 testing.create_thing_without_server which generates a mock interface.

title: str: A human-readable description of the Thing

_thing_server_interface: ThingServerInterface: Provide access to features of the server that this Thing is attached to.

_class_settings: ThingClassSettings

Specify LabThings features used (or incompatible with) this Thing. See documentation for ThingClassSettings for keys and their meanings. This property may be set using a dictionary literal, and most type checkers or IDEs should help catch incorrect keys or types. The keys and values are also validated when your class is defined.

This property should only be set during class definition: it should not be modified after the class is defined.

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

property logger: logging.Logger: A logger, named after this Thing. Use this logger if you wish to log messages from action or property code.

property properties: labthings_fastapi.properties.PropertyCollection

A mapping of names to PropertyInfo objects. This allows easy access to metadata, for example:

self.properties["myprop"].default

property settings: labthings_fastapi.properties.SettingCollection: A mapping of names to SettingInfo objects, similar to properties but providing setting-specific features.

property actions: labthings_fastapi.actions.ActionCollection: A mapping of names to ActionInfo objects that allows convenient access to metadata of each action.

property thing_state: collections.abc.Mapping: This should return a dictionary of metadata, which will be returned to any code requesting it through ThingServerInterface.get_thing_states.

get_current_invocation_logs() → list[LogRecord]

Get the log records for an on going action.

This is useful if an action wishes to save its logs alongside any data.

Note that only the last 1000 logs are returned so for long running tasks that log frequently this may want to be read periodically.

This will error if it is called outside an action invocation.

Returns:: a list of all logs from this action.
Raises:: RuntimeError – If the server cannot be retrieved. This should never happen.

lt.property(getter: Callable[[Owner], Value]) → FunctionalProperty[Owner, Value]

lt.property(*, default: Value, readonly: bool = False, use_global_lock: bool | None = None, **constraints: Any) → Value

lt.property(*, default_factory: Callable[[], Value], readonly: bool = False, use_global_lock: bool | None = None, **constraints: Any) → Value

This function may be used to define Properties either by decorating a function, or marking an attribute. Full documentation is available at labthings_fastapi.properties.property and a more in-depth discussion is available at Properties. This page focuses on the most frequently used examples.

To mark a class attribute with property you should define the attribute as shown below. Note that the type hint is required for LabThings to work properly.

class MyThing(lt.Thing):
    intprop: int = lt.property(default=0)
    """A simple read-write property"""

    readonly: int = lt.property(default=42, readonly=True)
    """This property may not be written to over HTTP"""

    listprop: list[int] = lt.property(default_factory=lambda: [1,2,3])
    """Mutable default values should be wrapped in a "factory function".""""

    positive: int = lt.property(default=1, gt=0)
    """Constraints may be used in the same way as for `pydantic.Field`"""

All the examples above are “data properties”. property can also define “functional properties” when used as a decorator:

class MyThing(lt.Thing):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._number = 0

    @lt.property
    def the_answer(self) -> int:
        """A read-only property."""
        return 42

    @lt.property
    def number(self) -> int:
        """A property that's got extra attributes set."""

    @number.setter
    def _set_number(self, value: int) -> None:
        self._number = value

    number.readonly = True  # This prevents it being written over HTTP
    number.constraints = {"ge": 0}  # This adds constraints to the schema
    number.default = 0  # This adds a default value to the documentation

For a full listing of attributes that may be modified, see DataProperty.

lt.setting(getter: Callable[[Owner], Value]) → FunctionalSetting[Owner, Value]
lt.setting(*, default: Value, readonly: bool = False, use_global_lock: bool | None = None, **constraints: Any) → Value
lt.setting(*, default_factory: Callable[[], Value], readonly: bool = False, use_global_lock: bool | None = None, **constraints: Any) → Value: A setting is a property that is saved to disk. It is defined in the same way as property but will be synchronised with the Thing‘s settings file. Full documentation is available at labthings_fastapi.properties.setting

@lt.action

@lt.action(use_global_lock: bool | None = None, **kwargs: Any)

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:: **kwargs – Keyword arguments are passed to the constructor of ActionDescriptor.

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

Declare a connection to another Thing in the same server.

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. A Mapping should be used if the slot supports multiple Things. For example:

class ThingA(lt.Thing): ...


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

    thing_a: ThingA = lt.thing_slot()
    multiple_things_a: Mapping[str, ThingA] = lt.thing_slot()

For more details, see the full API docs for thing_slot.

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.

@lt.endpoint(method: HTTPMethod, path: str | None = None, **kwargs: Any)

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:

The path is relative to the host Thing and will default to the name
of the method.
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 lt.ThingServer(config: ThingServerConfig, debug: bool = False)

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

Full documentation of how the class works is available at labthings_fastpi.server.ThingServer. Most of the attributes of ThingServer should not be accessed directly by Thing subclasses - instead they should use the ThingServerInterface for a cleaner way to access the server.

param config:

a ThingServerConfig instance (or compatible dictionary) setting the server’s configuration.

param debug:

sets the log level for Thing instances to DEBUG if it is set to True.

param **kwargs:

for backwards compatibility, keyword arguments are used to create the server configuration if config is missing. This raises a DeprecationWarning.

classmethod from_config(config: ThingServerConfig, debug: bool = False) → ThingServer

This method of creating a ThingServer is deprecated, as it is equivalent to the constructor.

Returns:

a ThingServer with the supplied configuration.
classmethod from_things(things: Mapping[str:ThingConfig | type[Thing] | str]) → ThingServer
This class method allows a server to be created by passing in a mapping of Thing names to Thing configurations. Thing configurations may be either a Thing subclass, or an import string (e.g. my.module:MyClass), or a ThingConfig instance (or compatible dictionary).
server = lt.ThingServer.from_things(
    {
        "my_thing": MyThingSubclass,
        "their_thing": "their.module:TheirThing",
        "configured_thing": {
            "cls": MyThingSubclass,
            "kwargs": {"a": 10},
        },
    }
)
Parameters:

things – a mapping of Thing names to Thing configurations.

**kwargs – additional keyword arguments are passed to ThingServerConfig.
property things: collections.abc.Mapping[str, Thing]

A read-only mapping of names to Thing instances, for every Thing attached to the server.

Returns:

a mapping of names to Thing instances.

property application_config: collections.abc.Mapping[str, Any] | None

Return the application configuration from the config file.

Returns:: The custom configuration as specified in the configuration file.

property api_prefix: str: A string that prefixes all URLs in the application. Endpoints associated with a Thing will appear under {api_prefix}/{thing_name}/. This is validated in ThingServerConfig and must either be empty, or start with / and not end with /.

property settings_folder

:type: str

The path to the folder used to store settings for each Thing. This will default to ./settings/ if not specified in the server’s configuration.

property debug: bool: Whether the server is in debug mode.

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

An interface for Things to interact with their server. This is available as Thing._thing_server_interface and should not normally be created except by the ThingServer.

start_async_task_soon(async_function: Callable[[Params], Awaitable[ReturnType]], *args: Any) → 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).

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

hold_global_lock(*, error_if_unavailable: bool = True): A context manager that holds the global lock. By default, an exception is raised if the global lock is not enabled. error_if_unavailable may be used to suppress that error, in which case the context manager silently does nothing if there is no global lock to acquire.

class lt.ThingClassSettings

Bases: typing_extensions.TypedDict

A typed dictionary to hold settings that determine how a Thing subclass interacts with LabThings. This is used to control Optional Features and updates.

validate_properties_on_set

Type:: bool
Default:: False

Whether properties should be validated against their model when set from Python. Properties are always validated when set over HTTP. By default, no validation is performed when they are set from Python. Setting this key to True will enable validation: this will become the default in the future.

class lt.ThingConfig(/, **data: Any)

Bases: pydantic.BaseModel

The information needed to add a Thing to a ThingServer. This is an alias of labthings_fastapi.server.config_model.ThingConfig

cls: Annotated[ImportString, WrapValidator(func=contain_import_errors, json_schema_input_type=PydanticUndefined)]

args: Sequence[Any]

kwargs: Mapping[str, Any]

thing_slots: Mapping[str, str | Iterable[str] | None]

class lt.ThingServerConfig(/, **data: Any)

Bases: pydantic.BaseModel

The configuration parameters for a ThingServer.

things: Mapping[Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1), _PydanticGeneralMetadata(pattern='^([a-zA-Z0-9\\-_]+)$')]), AfterValidator(func=check_reserved_thing_names)], ThingConfig | Annotated[ImportString, WrapValidator(func=contain_import_errors, json_schema_input_type=PydanticUndefined)]]

settings_folder: str | None

api_prefix: str

enable_global_lock: bool

application_config: dict[str, Any] | None

class lt.ThingClient

A client for a LabThings-FastAPI Thing, alias of labthings_fastapi.client.ThingClient

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.

classmethod from_url(thing_url: str, client: Client | None = None) → 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).

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

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

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.

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.

class lt.DataProperty: This is an alias of labthings_fastapi.properties.DataProperty but is not usually used: consider using property instead.

class lt.DataSetting: This is an alias of labthings_fastapi.properties.DataSetting but is not usually used: consider using property instead.

lt.outputs: This is an alias for labthings_fastapi.outputs and contains, for example MJPEGStream.

lt.blob: This module implements Blob input/output, which is the intended way to return files and binary data from LabThings. It is an alias for labthings_fastapi.outputs.blob.

lt.cli: The CLI module implements command-line functions, and is where the command-line script labthings-server is implemented. It is an alias for labthings_fastapi.server.cli.