labthings_fastapi.server

Code supporting the LabThings server.

LabThings wraps the fastapi.FastAPI application in a ThingServer, which provides the tools to serve and manage Thing instances.

See the LabThings-FastAPI tutorial for examples of how to set up a ThingServer.

Submodules

• labthings_fastapi.server.cli
• labthings_fastapi.server.config_model
• labthings_fastapi.server.fallback

Classes

ThingServer

Use FastAPI to serve Thing instances.

Package Contents

class labthings_fastapi.server.ThingServer(config: config_model.ThingServerConfig, *, debug: bool = False)

class labthings_fastapi.server.ThingServer(*, debug: bool = False, **kwargs: Any)

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.

The ThingServer is responsible for running the code in Thing instances, and making them available over the network. It should be configured by passing a ThingServerConfig object (or a dictionary that can be validated as a ThingServerConfig object).

For convenience and backwards compatibility, if config is None the keyword arguments will be passed to ThingServerConfig instead. Keyword arguments may not be used if the config argument is used, and may be removed in the future.

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.

Parameters:

• config – a ThingServerConfig object that configures the server, or something that may be converted to one.

• debug – ff True, set the log level for Thing instances to DEBUG.

• **kwargs – ff keyword arguments are supplied, they will be passed to the constructor of ThingServerConfig. This is not allowed if config is a ThingServerConfig object.

Raises:

• TypeError – if the value of config cannot be parsed as a ThingServerConfig.

• ValueError – if keyword arguments are supplied together with config.

startup_failure: dict | None = None

_debug = False

app

action_manager

message_broker

blocking_portal: anyio.from_thread.BlockingPortal | None = None

startup_status: dict[str, str | dict]

global_lock

_things

classmethod from_things(things: config_model.ThingsConfig, debug: bool = False, **kwargs: Any) → Self

Create a ThingServer using a dictionary of Thing subclasses.

In test and example code, it’s convenient to be able to pass server and Thing configurations as keyword arguments rather than a config model.

This convenience method will turn its keyword arguments into a server configuration and create a server based on it.

Parameters:

• things – A mapping of names to Thing configurations. These may be specified as a ThingConfig object, a Thing subclass, or an import string referencing a Thing subclass.

• debug – Whether to start the server in debug mode.

• **kwargs – Additional keyword arguments are passed to ThingServerConfig.

Returns:

a ThingServer instance.

classmethod from_config(config: config_model.ThingServerConfig, debug: bool = False) → Self

Create a ThingServer from a configuration model.

This is equivalent to ThingServer(config, debug=debug).

Parameters:

• config – The configuration parameters for the server.

• debug – If True, set the log level for Thing instances to DEBUG.

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.

_set_url_for_middleware() → None

Add middleware to support url_for in Pydantic models.

This middleware adds a request state variable that allows labthings_fastapi.server.URLFor instances to be serialised using FastAPI’s url_for function.

_add_exception_handlers() → None

Add exception handlers to the FastAPI application.

property debug: bool

Whether the server is in debug mode.

property settings_folder: str

The folder in which we will store Thing settings.

Raises:

RuntimeError – if there is no settings folder set. This should never happen, as it’s set in __init__.

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

A read-only 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.

This will either be empty, or start with a slash and not end with a slash. Validation is performed in ThingServerConfig.

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

Raises:

BaseException – Reraises any errors that are caught when calling __enter__ on each Thing. The error is also saved to self.startup_failure for post mortem, as otherwise uvicorn will swallow it and replace it with SystemExit(3) and no traceback.

_things_view_router() → fastapi.APIRouter

Create a router for the endpoint that shows the list of attached things.

Returns:

an APIRouter with the thing_descriptions endpoint.

serve(host: str = 'localhost', port: int = 5000) → None

Run the server in uvicorn.

This method will run the server from Python, using uvicorn.run. This is the most convenient way to run a LabThings server from Python, and is identical to what happens when it is run from the command line.

Parameters:

• host – The IP address or hostname on which to serve. By default, this is localhost which is only accessible from your computer. To serve over a network on all available IPv4 addresses, use "0.0.0.0".

• port – The port on which to serve. This defaults to 5000.

test_client() → collections.abc.Iterator[fastapi.testclient.TestClient]

A context manager to test out a server without binding to a port.

This context manager will start up the server and run an event loop, but instead of responding to requests on a network port, it uses fastapi.testclient.TestClient to simulate HTTP requests.

This is provided to simplify test code, and should not be used in production.

Yields:

a fastapi.testclient.TestClient to simulate HTTP requests.

Warning

Usually, a server is only started up and shut down once. Calling this method multiple times may have unexpected results.

As a rule, only ever use this method in your test suite.