API Reference¶
- exception labthings.ActionKilledException¶
Sibling of SystemExit, but specific to thread termination.
- class labthings.ClientEvent¶
An event-signaller object with per-client setting and waiting.
A client can be any Greenlet or native Thread. This can be used, for example, to signal to clients that new data is available
- clear() bool ¶
Clear frame event, once processed.
- set(timeout=5)¶
Signal that a new frame is available.
- Parameters
timeout – (Default value = 5)
- wait(timeout: int = 5)¶
Wait for the next data frame (invoked from each client’s thread).
- Parameters
timeout – int: (Default value = 5)
- class labthings.CompositeLock(locks, timeout: int = - 1)¶
Class that behaves like a
labthings.core.lock.StrictLock
, but allows multiple locks to be acquired and released.- Parameters
locks (list) – List of parent RLock objects
timeout (int) – Time in seconds acquisition will wait before raising an exception
- acquire(blocking: bool = True, timeout=<object object>)¶
- Parameters
blocking – (Default value = True)
timeout – (Default value = sentinel)
- class labthings.LabThing(app: Optional[flask.app.Flask] = None, id_: Optional[str] = None, prefix: str = '', title: str = '', description: str = '', version: str = '0.0.0', types: Optional[List[str]] = None, format_flask_exceptions: bool = True, external_links: bool = True, json_encoder=<class 'labthings.json.encoder.LabThingsJSONEncoder'>)¶
The main entry point for the application. You need to initialize it with a Flask Application:
>>> app = Flask(__name__) >>> labthing = labthings.LabThing(app)
Alternatively, you can use
init_app()
to set the Flask application after it has been constructed.- Parameters
app (flask.Flask) – the Flask application object
prefix (str) – Prefix all routes with a value, eg v1 or 2010-04-01
title (str) – Human-readable title of the Thing
description (str) – Human-readable description of the Thing
version (str) – Version number of the Thing
types (list of str) – List of Thing types, used by clients to filter discovered Things
format_flask_exceptions (bool) – JSON format all exception responses
external_links (bool) – Use external links in Thing Description where possible
json_encoder – JSON encoder class for the app
- add_component(component_object, component_name: str)¶
Add a component object to the LabThing, allowing it to be used by extensions and other views by name, rather than reference.
- Parameters
device_object – Component object
device_name – str: Component name, used by extensions to find the object
- add_root_link(view: Type[labthings.views.View], rel: str, kwargs=None, params=None)¶
- Parameters
view –
rel –
kwargs – (Default value = None)
params – (Default value = None)
- add_view(view: Type[labthings.views.View], *urls: str, endpoint: Optional[str] = None, **kwargs)¶
Adds a view to the api.
- Parameters
view – View class
urls (str) – one or more url routes to match for the resource, standard flask routing rules apply. Any url variables will be passed to the resource method as args.
endpoint (str) – endpoint name (defaults to
Resource.__name__()
Can be used to reference this route infields.Url
fieldskwargs – kwargs to be forwarded to the constructor of the view.
Additional keyword arguments not specified above will be passed as-is to
flask.Flask.add_url_rule()
.Examples:
labthing.add_view(HelloWorld, '/', '/hello') labthing.add_view(Foo, '/foo', endpoint="foo") labthing.add_view(FooSpecial, '/special/foo', endpoint="foo")
- property description: str¶
Human-readable description of the Thing
- emit(event_type: str, data: dict)¶
Find a matching event type if one exists, and emit some data to it
- Parameters
event_type – str:
data – dict:
- init_app(app)¶
Initialize this class with the given
flask.Flask
application. :param app: the Flask application or blueprint object- Examples::
labthing = LabThing() labthing.add_view(…) labthing.init_app(app)
- register_extension(extension_object: labthings.extensions.BaseExtension)¶
Add an extension to the LabThing. This will add API views and lifecycle functions from the extension to the LabThing
- Parameters
extension_object (labthings.extensions.BaseExtension) – Extension instance
- property safe_title: str¶
Lowercase title with no whitespace
- property title: str¶
Human-readable title of the Thing
- url_for(view: Type[labthings.views.View], **values)¶
Generates a URL to the given resource. Works like
flask.url_for()
.- Parameters
view –
values –
- property version: str¶
Version number of the Thing
- view(*urls: str, **kwargs)¶
Wraps a
labthings.View
class, adding it to the LabThing. Parameters are the same asadd_view()
.Example:
app = Flask(__name__) labthing = labthings.LabThing(app) @labthing.view('/properties/my_property') class Foo(labthings.views.PropertyView): schema = labthings.fields.String() def get(self): return 'Hello, World!'
- class labthings.Schema(*, only: Optional[Union[Sequence[str], Set[str]]] = None, exclude: Union[Sequence[str], Set[str]] = (), many: bool = False, context: Optional[Dict] = None, load_only: Union[Sequence[str], Set[str]] = (), dump_only: Union[Sequence[str], Set[str]] = (), partial: Union[bool, Sequence[str], Set[str]] = False, unknown: Optional[str] = None)¶
Base schema class with which to define custom schemas.
Example usage:
import datetime as dt from dataclasses import dataclass from marshmallow import Schema, fields @dataclass class Album: title: str release_date: dt.date class AlbumSchema(Schema): title = fields.Str() release_date = fields.Date() album = Album("Beggars Banquet", dt.date(1968, 12, 6)) schema = AlbumSchema() data = schema.dump(album) data # {'release_date': '1968-12-06', 'title': 'Beggars Banquet'}
- Parameters
only – Whitelist of the declared fields to select when instantiating the Schema. If None, all fields are used. Nested fields can be represented with dot delimiters.
exclude – Blacklist of the declared fields to exclude when instantiating the Schema. If a field appears in both only and exclude, it is not used. Nested fields can be represented with dot delimiters.
many – Should be set to True if
obj
is a collection so that the object will be serialized to a list.context – Optional context passed to
fields.Method
andfields.Function
fields.load_only – Fields to skip during serialization (write-only fields)
dump_only – Fields to skip during deserialization (read-only fields)
partial – Whether to ignore missing fields and not require any fields declared. Propagates down to
Nested
fields as well. If its value is an iterable, only missing fields listed in that iterable will be ignored. Use dot delimiters to specify nested fields.unknown – Whether to exclude, include, or raise an error for unknown fields in the data. Use EXCLUDE, INCLUDE or RAISE.
Changed in version 3.0.0: prefix parameter removed.
Changed in version 2.0.0: __validators__, __preprocessors__, and __data_handlers__ are removed in favor of marshmallow.decorators.validates_schema, marshmallow.decorators.pre_load and marshmallow.decorators.post_dump. __accessor__ and __error_handler__ are deprecated. Implement the handle_error and get_attribute methods instead.
- class Meta¶
Options object for a Schema.
Example usage:
class Meta: fields = ("id", "email", "date_created") exclude = ("password", "secret_attribute")
Available options:
fields
: Tuple or list of fields to include in the serialized result.additional
: Tuple or list of fields to include in addition to theexplicitly declared fields.
additional
andfields
are mutually-exclusive options.
include
: Dictionary of additional fields to include in the schema. It isusually better to define fields as class variables, but you may need to use this option, e.g., if your fields are Python keywords. May be an OrderedDict.
exclude
: Tuple or list of fields to exclude in the serialized result.Nested fields can be represented with dot delimiters.
dateformat
: Default format for Date <fields.Date> fields.datetimeformat
: Default format for DateTime <fields.DateTime> fields.timeformat
: Default format for Time <fields.Time> fields.render_module
: Module to use for loads <Schema.loads> and dumps <Schema.dumps>.Defaults to json from the standard library.
ordered
: If True, order serialization output according to theorder in which fields were declared. Output of Schema.dump will be a collections.OrderedDict.
index_errors
: If True, errors dictionaries will include the indexof invalid items in a collection.
load_only
: Tuple or list of fields to exclude from serialized results.dump_only
: Tuple or list of fields to exclude from deserializationunknown
: Whether to exclude, include, or raise an error for unknownfields in the data. Use EXCLUDE, INCLUDE or RAISE.
register
: Whether to register the Schema with marshmallow’s internalclass registry. Must be True if you intend to refer to this Schema by class name in Nested fields. Only set this to False when memory usage is critical. Defaults to True.
- OPTIONS_CLASS¶
alias of
marshmallow.schema.SchemaOpts
- dump(obj: Any, *, many: Optional[bool] = None)¶
Serialize an object to native Python data types according to this Schema’s fields.
- Parameters
obj – The object to serialize.
many – Whether to serialize obj as a collection. If None, the value for self.many is used.
- Returns
Serialized data
New in version 1.0.0.
Changed in version 3.0.0b7: This method returns the serialized data rather than a
(data, errors)
duple. AValidationError
is raised ifobj
is invalid.Changed in version 3.0.0rc9: Validation no longer occurs upon serialization.
- dumps(obj: Any, *args, many: Optional[bool] = None, **kwargs)¶
Same as
dump()
, except return a JSON-encoded string.- Parameters
obj – The object to serialize.
many – Whether to serialize obj as a collection. If None, the value for self.many is used.
- Returns
A
json
string
New in version 1.0.0.
Changed in version 3.0.0b7: This method returns the serialized data rather than a
(data, errors)
duple. AValidationError
is raised ifobj
is invalid.
- error_messages: Dict[str, str] = {}¶
Overrides for default schema-level error messages
- fields: Dict[str, marshmallow.fields.Field]¶
Dictionary mapping field_names ->
Field
objects
- classmethod from_dict(fields: Dict[str, Union[marshmallow.fields.Field, type]], *, name: str = 'GeneratedSchema') type ¶
Generate a Schema class given a dictionary of fields.
from marshmallow import Schema, fields PersonSchema = Schema.from_dict({"name": fields.Str()}) print(PersonSchema().load({"name": "David"})) # => {'name': 'David'}
Generated schemas are not added to the class registry and therefore cannot be referred to by name in Nested fields.
- Parameters
fields (dict) – Dictionary mapping field names to field instances.
name (str) – Optional name for the class, which will appear in the
repr
for the class.
New in version 3.0.0.
- get_attribute(obj: Any, attr: str, default: Any)¶
Defines how to pull values from an object to serialize.
New in version 2.0.0.
Changed in version 3.0.0a1: Changed position of
obj
andattr
.
- handle_error(error: marshmallow.exceptions.ValidationError, data: Any, *, many: bool, **kwargs)¶
Custom error handler function for the schema.
- Parameters
error – The ValidationError raised during (de)serialization.
data – The original input data.
many – Value of
many
on dump or load.partial – Value of
partial
on load.
New in version 2.0.0.
Changed in version 3.0.0rc9: Receives many and partial (on deserialization) as keyword arguments.
- load(data: Union[Mapping[str, Any], Iterable[Mapping[str, Any]]], *, many: Optional[bool] = None, partial: Optional[Union[bool, Sequence[str], Set[str]]] = None, unknown: Optional[str] = None)¶
Deserialize a data structure to an object defined by this Schema’s fields.
- Parameters
data – The data to deserialize.
many – Whether to deserialize data as a collection. If None, the value for self.many is used.
partial – Whether to ignore missing fields and not require any fields declared. Propagates down to
Nested
fields as well. If its value is an iterable, only missing fields listed in that iterable will be ignored. Use dot delimiters to specify nested fields.unknown – Whether to exclude, include, or raise an error for unknown fields in the data. Use EXCLUDE, INCLUDE or RAISE. If None, the value for self.unknown is used.
- Returns
Deserialized data
New in version 1.0.0.
Changed in version 3.0.0b7: This method returns the deserialized data rather than a
(data, errors)
duple. AValidationError
is raised if invalid data are passed.
- loads(json_data: str, *, many: Optional[bool] = None, partial: Optional[Union[bool, Sequence[str], Set[str]]] = None, unknown: Optional[str] = None, **kwargs)¶
Same as
load()
, except it takes a JSON string as input.- Parameters
json_data – A JSON string of the data to deserialize.
many – Whether to deserialize obj as a collection. If None, the value for self.many is used.
partial – Whether to ignore missing fields and not require any fields declared. Propagates down to
Nested
fields as well. If its value is an iterable, only missing fields listed in that iterable will be ignored. Use dot delimiters to specify nested fields.unknown – Whether to exclude, include, or raise an error for unknown fields in the data. Use EXCLUDE, INCLUDE or RAISE. If None, the value for self.unknown is used.
- Returns
Deserialized data
New in version 1.0.0.
Changed in version 3.0.0b7: This method returns the deserialized data rather than a
(data, errors)
duple. AValidationError
is raised if invalid data are passed.
- on_bind_field(field_name: str, field_obj: marshmallow.fields.Field) None ¶
Hook to modify a field when it is bound to the Schema.
No-op by default.
- validate(data: Mapping, *, many: Optional[bool] = None, partial: Optional[Union[bool, Sequence[str], Set[str]]] = None) Dict[str, List[str]] ¶
Validate data against the schema, returning a dictionary of validation errors.
- Parameters
data – The data to validate.
many – Whether to validate data as a collection. If None, the value for self.many is used.
partial – Whether to ignore missing fields and not require any fields declared. Propagates down to
Nested
fields as well. If its value is an iterable, only missing fields listed in that iterable will be ignored. Use dot delimiters to specify nested fields.
- Returns
A dictionary of validation errors.
New in version 1.1.0.
- class labthings.Server(app, host='0.0.0.0', port=7485, debug=False, zeroconf=True)¶
Combined WSGI+mDNS server.
- Parameters
host (string) – Host IP address. Defaults to 0.0.0.0.
port (int) – Host port. Defaults to 7485.
debug (bool) – Enable server debug mode. Defaults to False.
zeroconf (bool) – Enable the zeroconf (mDNS) server. Defaults to True.
- run(host=None, port=None, debug=None, zeroconf=None)¶
Starts the server allowing for runtime parameters. Designed to immitate the old Flask app.run style of starting an app
- Parameters
host (string) – Host IP address. Defaults to 0.0.0.0.
port (int) – Host port. Defaults to 7485.
debug (bool) – Enable server debug mode. Defaults to False.
zeroconf (bool) – Enable the zeroconf (mDNS) server. Defaults to True.
- start()¶
Start the server and register mDNS records
- class labthings.StrictLock(timeout: int = - 1, name: Optional[str] = None)¶
Class that behaves like a Python RLock, but with stricter timeout conditions and custom exceptions.
- Parameters
timeout (int) – Time in seconds acquisition will wait before raising an exception
- acquire(blocking: bool = True, timeout=<object object>, _strict: bool = True)¶
- Parameters
blocking – (Default value = True)
timeout – (Default value = sentinel)
_strict – (Default value = True)
- labthings.create_app(import_name, prefix: str = '', title: str = '', description: str = '', types: Optional[list] = None, version: str = '0.0.0', external_links: bool = True, handle_errors: bool = True, handle_cors: bool = True, flask_kwargs: Optional[dict] = None)¶
Quick-create a LabThings-enabled Flask app
- Parameters
import_name – Flask import name. Usually
__name__
.prefix (str) – URL prefix for all LabThings views. Defaults to “/api”.
title (str) – Title/name of the LabThings Thing.
description (str) – Brief description of the LabThings Thing.
version (str) – Version number/code of the Thing. Defaults to “0.0.0”.
handle_errors (bool) – Use the LabThings error handler, to JSON format internal exceptions. Defaults to True.
handle_cors (bool) – Automatically enable CORS on all LabThings views. Defaults to True.
flask_kwargs (dict) – Keyword arguments to pass to the Flask instance.
prefix – str: (Default value = “”)
title – str: (Default value = “”)
description – str: (Default value = “”)
types – list: (Default value = None)
version – str: (Default value = “0.0.0”)
external_links – bool: Use external links in Thing Description where possible
handle_errors – bool: (Default value = True)
handle_cors – bool: (Default value = True)
flask_kwargs – dict: (Default value = None)
- Returns
(Flask app object, LabThings object)
- labthings.current_action()¶
Return the ActionThread instance in which the caller is currently running.
If this function is called from outside an ActionThread, it will return None.
- Returns
labthings.actions.ActionThread
– Currently running ActionThread.
- labthings.current_labthing(app=None)¶
The LabThing instance handling current requests.
Searches for a valid LabThing extension attached to the current Flask context.
- Parameters
app – (Default value = None)
- labthings.find_component(component_name: str, labthing_instance=None)¶
Find a particular LabThings Component registered to a LabThing instance
- Parameters
component_name (str) – Fully qualified name of the component
labthing_instance (optional) – LabThing instance to search for the component. Defaults to current_labthing.
- Returns
Python object registered as a component, or None if not found
- labthings.find_extension(extension_name: str, labthing_instance=None)¶
Find a particular LabThings Extension registered to a LabThing instance
- Parameters
extension_name (str) – Fully qualified name of the extension
labthing_instance (optional) – LabThing instance to search for the extension. Defaults to current_labthing.
- Returns
LabThings Extension object, or None if not found
- labthings.registered_components(labthing_instance=None)¶
Find all LabThings Components registered to a LabThing instance
- Parameters
labthing_instance (optional) – LabThing instance to search for extensions. Defaults to current_labthing.
- Returns
Python objects registered as LabThings components
- Return type
list
- labthings.registered_extensions(labthing_instance=None)¶
Find all LabThings Extensions registered to a LabThing instance
- Parameters
labthing_instance (optional) – LabThing instance to search for extensions. Defaults to current_labthing.
- Returns
LabThing Extension objects
- Return type
list
- labthings.update_action_data(data: dict)¶
Update the data of the ActionThread in which the caller is currently running.
If this function is called from outside an ActionThread, it will do nothing.
- Parameters
data – dict: Action data dictionary
- labthings.update_action_progress(progress: int)¶
Update the progress of the ActionThread in which the caller is currently running.
If this function is called from outside an ActionThread, it will do nothing.
- Parameters
progress – int: Action progress, in percent (0-100)