347 lines
11 KiB
Python
347 lines
11 KiB
Python
from flask import url_for, jsonify
|
|
from apispec import APISpec
|
|
from apispec.ext.marshmallow import MarshmallowPlugin
|
|
|
|
from .plugins import BasePlugin
|
|
from .views.plugins import PluginListResource
|
|
from .views.tasks import TaskList, TaskResource
|
|
|
|
from .spec import view2path
|
|
|
|
from .utilities import description_from_view
|
|
|
|
from openflexure_microscope.common.labthings_core.utilities import get_docstring
|
|
from .exceptions import JSONExceptionHandler
|
|
|
|
from . import EXTENSION_NAME
|
|
|
|
import logging
|
|
|
|
|
|
class LabThing(object):
|
|
def __init__(
|
|
self,
|
|
app=None,
|
|
prefix: str = "",
|
|
title: str = "",
|
|
description: str = "",
|
|
version: str = "0.0.0",
|
|
handle_errors: bool = True,
|
|
):
|
|
self.app = app
|
|
|
|
self.devices = {}
|
|
|
|
self.plugins = {}
|
|
|
|
self.resources = []
|
|
self.properties = {}
|
|
self.actions = {}
|
|
|
|
self.endpoints = set()
|
|
|
|
self.url_prefix = prefix
|
|
self._description = description
|
|
self._title = title
|
|
self._version = version
|
|
|
|
if handle_errors:
|
|
self.error_handler = JSONExceptionHandler()
|
|
else:
|
|
self.error_handler = None
|
|
|
|
self.spec = APISpec(
|
|
title=self.title,
|
|
version=self.version,
|
|
openapi_version="3.0.2",
|
|
plugins=[MarshmallowPlugin()],
|
|
)
|
|
|
|
if app is not None:
|
|
self.init_app(app)
|
|
|
|
@property
|
|
def description(self,):
|
|
return self._description
|
|
|
|
@description.setter
|
|
def description(self, description: str):
|
|
self._description = description
|
|
self.spec.description = description
|
|
|
|
@property
|
|
def title(self,):
|
|
return self._title
|
|
|
|
@title.setter
|
|
def title(self, title: str):
|
|
self._title = title
|
|
self.spec.title = title
|
|
|
|
@property
|
|
def version(self,):
|
|
return str(self._version)
|
|
|
|
@version.setter
|
|
def version(self, version: str):
|
|
self._version = version
|
|
self.spec.version = version
|
|
|
|
### Flask stuff
|
|
|
|
def init_app(self, app):
|
|
app.teardown_appcontext(self.teardown)
|
|
|
|
# Register Flask extension
|
|
app.extensions = getattr(app, "extensions", {})
|
|
app.extensions[EXTENSION_NAME] = self
|
|
|
|
# Register error handler if one exists
|
|
if self.error_handler:
|
|
self.error_handler.init_app(self.app)
|
|
|
|
# Add resources, if registered before tying to a Flask app
|
|
if len(self.resources) > 0:
|
|
for resource, urls, endpoint, kwargs in self.resources:
|
|
self._register_view(app, resource, *urls, endpoint=endpoint, **kwargs)
|
|
|
|
# Create base routes
|
|
self._create_base_routes()
|
|
|
|
def teardown(self, exception):
|
|
pass
|
|
|
|
def _create_base_routes(self):
|
|
# Add root representation
|
|
self.app.add_url_rule(self._complete_url("/", ""), "rootrep", self.rootrep)
|
|
# Add thing description
|
|
self.app.add_url_rule(self._complete_url("/td", ""), "td", self.td)
|
|
# Add swagger spec
|
|
self.app.add_url_rule(
|
|
self._complete_url("/swagger", ""), "swagger", self.swagger
|
|
)
|
|
|
|
# Add plugin overview
|
|
self.add_resource(PluginListResource, "/plugins")
|
|
self.register_property(PluginListResource)
|
|
# Add task routes
|
|
self.add_resource(TaskList, "/tasks")
|
|
self.register_property(TaskList)
|
|
self.add_resource(TaskResource, "/tasks/<id>")
|
|
|
|
### Device stuff
|
|
|
|
def register_device(self, device_object, device_name: str):
|
|
self.devices[device_name] = device_object
|
|
|
|
### Plugin stuff
|
|
|
|
def register_plugin(self, plugin_object):
|
|
if isinstance(plugin_object, BasePlugin):
|
|
self.plugins[plugin_object.name] = plugin_object
|
|
else:
|
|
raise TypeError("Plugin object must be an instance of BasePlugin")
|
|
|
|
for plugin_view_id, plugin_view in plugin_object.views.items():
|
|
# Add route to the plugins blueprint
|
|
self.add_resource(
|
|
plugin_view["view"],
|
|
"/plugins" + plugin_view["rule"],
|
|
**plugin_view["kwargs"],
|
|
)
|
|
|
|
for prop in plugin_object.properties:
|
|
self.register_property(prop)
|
|
|
|
for action in plugin_object.actions:
|
|
self.register_action(action)
|
|
|
|
### Resource stuff
|
|
|
|
def _complete_url(self, url_part, registration_prefix):
|
|
"""This method is used to defer the construction of the final url in
|
|
the case that the Api is created with a Blueprint.
|
|
:param url_part: The part of the url the endpoint is registered with
|
|
:param registration_prefix: The part of the url contributed by the
|
|
blueprint. Generally speaking, BlueprintSetupState.url_prefix
|
|
"""
|
|
parts = [registration_prefix, self.url_prefix, url_part]
|
|
return "".join([part for part in parts if part])
|
|
|
|
def register_property(self, resource):
|
|
if hasattr(resource, "endpoint"):
|
|
self.properties[resource.endpoint] = resource
|
|
else:
|
|
raise RuntimeError(
|
|
f"Resource {resource} has not yet been added. Cannot set as a property."
|
|
)
|
|
|
|
def register_action(self, resource):
|
|
if hasattr(resource, "endpoint"):
|
|
self.actions[resource.endpoint] = resource
|
|
else:
|
|
raise RuntimeError(
|
|
f"Resource {resource} has not yet been added. Cannot set as an action."
|
|
)
|
|
|
|
def add_resource(self, resource, *urls, endpoint=None, **kwargs):
|
|
"""Adds a resource to the api.
|
|
:param resource: the class name of your resource
|
|
:type resource: :class:`Type[Resource]`
|
|
:param urls: 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.
|
|
:type urls: str
|
|
:param endpoint: endpoint name (defaults to :meth:`Resource.__name__`
|
|
Can be used to reference this route in :class:`fields.Url` fields
|
|
:type endpoint: str
|
|
:param resource_class_args: args to be forwarded to the constructor of
|
|
the resource.
|
|
:type resource_class_args: tuple
|
|
:param resource_class_kwargs: kwargs to be forwarded to the constructor
|
|
of the resource.
|
|
:type resource_class_kwargs: dict
|
|
Additional keyword arguments not specified above will be passed as-is
|
|
to :meth:`flask.Flask.add_url_rule`.
|
|
Examples::
|
|
api.add_resource(HelloWorld, '/', '/hello')
|
|
api.add_resource(Foo, '/foo', endpoint="foo")
|
|
api.add_resource(FooSpecial, '/special/foo', endpoint="foo")
|
|
"""
|
|
endpoint = endpoint or resource.__name__.lower()
|
|
|
|
logging.debug(f"{endpoint}: {type(resource)}")
|
|
|
|
if self.app is not None:
|
|
self._register_view(self.app, resource, *urls, endpoint=endpoint, **kwargs)
|
|
|
|
self.resources.append((resource, urls, endpoint, kwargs))
|
|
|
|
def resource(self, *urls, **kwargs):
|
|
"""Wraps a :class:`~flask_restful.Resource` class, adding it to the
|
|
api. Parameters are the same as :meth:`~flask_restful.Api.add_resource`.
|
|
Example::
|
|
app = Flask(__name__)
|
|
api = restful.Api(app)
|
|
@api.resource('/foo')
|
|
class Foo(Resource):
|
|
def get(self):
|
|
return 'Hello, World!'
|
|
"""
|
|
|
|
def decorator(cls):
|
|
self.add_resource(cls, *urls, **kwargs)
|
|
return cls
|
|
|
|
return decorator
|
|
|
|
def _register_view(self, app, resource, *urls, endpoint=None, **kwargs):
|
|
endpoint = endpoint or resource.__name__.lower()
|
|
self.endpoints.add(endpoint)
|
|
resource_class_args = kwargs.pop("resource_class_args", ())
|
|
resource_class_kwargs = kwargs.pop("resource_class_kwargs", {})
|
|
|
|
# NOTE: 'view_functions' is cleaned up from Blueprint class in Flask 1.0
|
|
if endpoint in getattr(app, "view_functions", {}):
|
|
previous_view_class = app.view_functions[endpoint].__dict__["view_class"]
|
|
|
|
# if you override the endpoint with a different class, avoid the collision by raising an exception
|
|
if previous_view_class != resource:
|
|
raise ValueError(
|
|
"This endpoint (%s) is already set to the class %s."
|
|
% (endpoint, previous_view_class.__name__)
|
|
)
|
|
|
|
resource.endpoint = endpoint
|
|
resource_func = resource.as_view(
|
|
endpoint, *resource_class_args, **resource_class_kwargs
|
|
)
|
|
|
|
for url in urls:
|
|
# If we've got no Blueprint, just build a url with no prefix
|
|
rule = self._complete_url(url, "")
|
|
# Add the url to the application or blueprint
|
|
app.add_url_rule(rule, view_func=resource_func, **kwargs)
|
|
# Add the resource to our API spec
|
|
self.spec.path(**view2path(rule, resource, self.spec))
|
|
|
|
### Utilities
|
|
|
|
def url_for(self, resource, **values):
|
|
"""Generates a URL to the given resource.
|
|
Works like :func:`flask.url_for`."""
|
|
endpoint = resource.endpoint
|
|
return url_for(endpoint, **values)
|
|
|
|
def owns_endpoint(self, endpoint):
|
|
return endpoint in self.endpoints
|
|
|
|
### Description
|
|
|
|
def td(self):
|
|
"""
|
|
W3C-style Thing Description
|
|
"""
|
|
props = {}
|
|
for key, prop in self.properties.items():
|
|
props[key] = {}
|
|
props[key]["title"] = prop.__name__
|
|
props[key]["description"] = get_docstring(prop)
|
|
props[key]["links"] = [{"href": self.url_for(prop, _external=True)}]
|
|
|
|
actions = {}
|
|
for key, prop in self.actions.items():
|
|
actions[key] = {}
|
|
actions[key]["title"] = prop.__name__
|
|
actions[key]["description"] = get_docstring(prop)
|
|
actions[key]["links"] = [{"href": self.url_for(prop, _external=True)}]
|
|
|
|
td = {
|
|
"id": url_for("td", _external=True),
|
|
"title": self.title,
|
|
"description": self.description,
|
|
"properties": props,
|
|
"actions": actions,
|
|
}
|
|
|
|
return jsonify(td)
|
|
|
|
def rootrep(self):
|
|
"""
|
|
Root representation
|
|
"""
|
|
# TODO: Allow custom root representations
|
|
|
|
rr = {
|
|
"id": url_for("rootrep", _external=True),
|
|
"title": self.title,
|
|
"description": self.description,
|
|
"links": {
|
|
"thingDescription": {
|
|
"href": url_for("td", _external=True),
|
|
"description": get_docstring(self.td),
|
|
"methods": ["GET"],
|
|
},
|
|
"swagger": {
|
|
"href": url_for("swagger", _external=True),
|
|
"description": get_docstring(self.swagger),
|
|
"methods": ["GET"],
|
|
},
|
|
"plugins": {
|
|
"href": self.url_for(PluginListResource, _external=True),
|
|
**description_from_view(PluginListResource),
|
|
},
|
|
"tasks": {
|
|
"href": self.url_for(TaskList, _external=True),
|
|
**description_from_view(TaskList),
|
|
},
|
|
},
|
|
}
|
|
|
|
return jsonify(rr)
|
|
|
|
def swagger(self):
|
|
"""
|
|
OpenAPI v3 documentation
|
|
"""
|
|
return jsonify(self.spec.to_dict())
|