server.py

# -*- coding: utf-8 -*-
import asyncio
import functools
import json
import os
import traceback
from typing import Dict, Union

import rdflib
from aiohttp import web
from aiohttp.web import middleware
from aiohttp.web_request import Request
from multidict import MultiDict
from wzl.mqtt import MQTTPublisher

from .error import ServerException
from ..soil.component import Component
from ..soil.element import Element
from ..soil.error import InvokationException, ReadOnlyException, ChildNotFoundException
from ..soil.function import Function
from ..soil.measurement import Measurement
from ..soil.parameter import Parameter
from ..soil.semantics import Semantics, Namespaces
from ..soil.stream import StreamScheduler
from ..soil.variable import Variable
from ..utils import root_logger
from ..utils import serialize
from ..utils.constants import BASE_UUID_PATTERN, HTTP_GET, HTTP_OPTIONS
from ..utils.error import DeviceException, UserException, PathResolutionException
from ..utils.resources import ResourceType

logger = root_logger.get(__name__)


@middleware
async def cors(request, handler):
    logger.info("CORS Middleware handles request from {}".format(request.url))
    logger.debug('Request Headers: {}'.format(request.headers))
    response = web.Response()
    # check if the request is a preflight request
    if 'Access-Control-Request-Method' in request.headers and request.headers['Access-Control-Request-Method'] in [
        "POST", "PATCH"]:
        response.headers.update({'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, PATCH',
                                 'Access-Control-Allow-Headers': request.headers['Access-Control-Request-Headers']})
        response.headers.update({'Access-Control-Allow-Origin': request.headers['Origin']})
        logger.debug('Preflight Response Headers :{}'.format(response.headers))
        return response
    try:
        response = await handler(request)
    except Exception as e:
        logger.error(traceback.format_exc())
        response = web.json_response({"description": str(e)}, status=500)
        logger.error(["[CORS] {} at {}".format(str(e), request.url)])
    finally:
        response.headers.update({'Access-Control-Allow-Origin': request.headers.get('Origin', "*")})
        logger.debug('Response Headers :{}'.format(response.headers))
        return response


class HTTPServer(object):
    """Provides a web application as of the aiohttp-library.

    """

    def __init__(self, loop: asyncio.AbstractEventLoop, host: str, port: int, model: Component,
                 dataformat: str = 'json', legacy_mode=False, scheduler: StreamScheduler = None,
                 publisher: MQTTPublisher = None, profiles_path: str = None):
        """Constructor

        Args:
            loop: The asyncio-event-loop used to execute the server.
            host: Hostname of the server, i.e. a URL such as 'localhost'.
            port: Port the server should run at.
            model: The root component of the SOIL model, should be initialized via Component.load(...)
            dataformat: String specifying the dataformat of the responses of the server, either 'json' (default) or 'xml'.
            legacy_mode: If true, the datatypes are serialized to "bool" and "float" (instead of "boolean" and "float").
            scheduler: Stream handler. Is required to hand over jobs of streams and events for dynamic components, if created at runtime.
        """
        if dataformat not in ['json', 'xml']:
            raise ValueError('Dataformat must be one of "json" or "xml".')

        self.loop = loop
        self.host = host
        self.port = port
        self.root = model
        self._dataformat = dataformat
        self._legacy_mode = legacy_mode
        self._scheduler = scheduler
        self._publisher = publisher
        self._profiles_path = profiles_path

        self.app = web.Application(loop=self.loop, middlewares=[cors])

        # define two routes for each request to make the 'objects' part optional
        self.app.router.add_get(r'/objects{uuids:(/' + BASE_UUID_PATTERN + r')*/?}', self.get)
        self.app.router.add_get(r'/{uuids:(' + BASE_UUID_PATTERN + r'($|/))*}', self.get)
        self.app.router.add_post(r'/objects{uuids:(/' + BASE_UUID_PATTERN + r')*/?}', self.post)
        self.app.router.add_post(r'/{uuids:(' + BASE_UUID_PATTERN + r'($|/))*}', self.post)
        self.app.router.add_delete(r'/objects{uuids:(/' + BASE_UUID_PATTERN + r')*/?}', self.delete)
        self.app.router.add_delete(r'/{uuids:(' + BASE_UUID_PATTERN + r'($|/))*}', self.delete)
        self.app.router.add_route('OPTIONS', r'/objects{uuids:(/' + BASE_UUID_PATTERN + r')*/?}', self.options)
        self.app.router.add_route('OPTIONS', r'/{uuids:(' + BASE_UUID_PATTERN + r'($|/))*}', self.options)
        self.app.router.add_put(r'/objects{uuids:(/' + BASE_UUID_PATTERN + r')*/?}', self.put)
        self.app.router.add_put(r'/{uuids:(' + BASE_UUID_PATTERN + r'($|/))*}', self.put)
        self.app.router.add_patch(r'/objects{uuids:(/' + BASE_UUID_PATTERN + r')*/?}', self.patch)
        self.app.router.add_patch(r'/{uuids:(' + BASE_UUID_PATTERN + r'($|/))*}', self.patch)
        web.run_app(self.app, host=self.host, port=self.port, loop=loop)
        logger.info('HTTP-Server serving on {}:{}'.format(host, port))

    @staticmethod
    def analyze_request_url(request: Request) -> ResourceType:
        assert request.url.parts[0] == '/'

        url_parts = request.url.parts
        if url_parts[-1] == '':
            url_parts = url_parts[:-1]

        if len(url_parts) == 3 and url_parts[1] == Semantics.prefix:
            return ResourceType.profile if 'Profile' in url_parts[-1] else ResourceType.metadata

        if request.url.path == '/' or url_parts[-1][:3] in ['COM', 'FUN', 'PAR', 'MEA', 'ARG', 'RET']:
            return ResourceType.element

        return ResourceType.metadata

    @staticmethod
    def parse_uuids(request: Request):
        """Splits the request URL to extract the FQID of the targeted element of the SOIL-Interface.

        Args:
            request:

        Returns:

        """
        uuids = request.match_info.get('uuids', 'uuids')
        uuid_list = uuids.split('/')
        while '' in uuid_list:
            uuid_list.remove('')
        return uuid_list

    def _filter_query(self, query: MultiDict):
        queried_attributes = []
        for key in query:
            if key in ['uuid', 'name', 'description', 'datatype', 'range', 'value', 'constant', 'timestamp',
                       'dimension', 'unit', 'covariance', 'label', 'children', 'arguments', 'returns', 'ontology']:
                queried_attributes += [key]
        return queried_attributes

    def prepare_response(self, body: Union[Dict, rdflib.Graph], element: Element, status: int = 200,
                         query: MultiDict = None, semantic: bool = False):
        dataformat = self._dataformat
        if query is not None and 'format' in query and query['format'] in ['json', 'xml', 'turtle']:
            dataformat = query['format']
        if query is not None and 'semantic' in query and query['semantic'] in ['data', 'metadata', 'profile']:
            semantic = True

        if dataformat == 'json':
            if semantic:
                assert isinstance(body, rdflib.Graph)
                # rdflib serialization returns a string, so we need to parse it as plain json again to return it properly
                body = json.loads(body.serialize(format='json-ld'))
            return web.json_response(body, status=status)
        elif dataformat == 'xml':
            if semantic:
                assert isinstance(body, rdflib.Graph)
                xml = body.serialize(format='xml')
            else:
                if element is not None and 200 <= status <= 300:
                    root = ''
                    if isinstance(element, Component):
                        root = 'component'
                    elif isinstance(element, Function):
                        root = 'function'
                    elif isinstance(element, Measurement):
                        root = 'measurement'
                    elif isinstance(element, Parameter):
                        root = 'parameter'
                else:
                    root = 'error'
                xml = serialize.to_xml(root, body)
            return web.Response(text=xml, status=status, content_type='application/xml')
        elif dataformat == 'turtle':
            if semantic:
                assert isinstance(body, rdflib.Graph)
                text = body.serialize(format='turtle')
            else:
                return web.Response(text='Can not serialize non semantic information to Turtle', status=400,
                                    content_type='text/plain')
            return web.Response(text=text, status=status, content_type=' text/plain')

    async def get(self, request):
        logger.info("GET Request from {}".format(request.url))
        logger.debug('Request: {}'.format(request))
        logger.debug('Query Parameters: {}'.format(request.query_string))

        resource_type = HTTPServer.analyze_request_url(request)
        keys = self._filter_query(request.query)

        try:
            if resource_type == ResourceType.profile:
                if self._profiles_path is None:
                    raise UserException('Can\'t return requested metadata profile, as no profiles have been created for this sensor service.')

                profilename = request.url.parts[-2] if request.url.parts[-1] == '' else request.url.parts[-1]

                if len(profilename) > 12 and profilename[-12:-7] == 'Range':
                    filename = profilename.replace('RangeProfile', '.shacl.ttl')
                else:
                    filename = profilename.replace('Profile', '.shacl.ttl')

                profile_path = os.path.join(self._profiles_path, filename)
                response = rdflib.Graph()
                response.parse(profile_path)
                response.add(
                    (rdflib.URIRef(Semantics.namespace[profilename]), Namespaces.dcterms.license, Semantics.profile_license))
                item, status = None, 200

            elif resource_type == ResourceType.metadata or resource_type == ResourceType.data:
                semantic_name = request.url.parts[-2] if request.url.parts[-1] == '' else request.url.parts[-1]

                item, resource_type = self.root.resolve_semantic_path(semantic_name)
                recursive = request.query is not None and 'all' in request.query
                response = item.serialize_semantics(resource_type, recursive)
            else:
                assert resource_type == ResourceType.element
                uuids = HTTPServer.parse_uuids(request)

                if request.query is not None and 'semantic' in request.query and request.query[
                    'semantic'] in ResourceType.semantic_resources:
                    resource_type = ResourceType.from_string(request.query['semantic'])

                try:
                    item = self.root[uuids]

                    if resource_type.is_semantic():
                        recursive = request.query is not None and 'all' in request.query
                        response = item.serialize_semantics(resource_type, recursive)
                    else:
                        response = item.serialize(keys, self._legacy_mode, HTTP_GET)
                except KeyError as e:
                    logger.error(traceback.format_exc())
                    response = {'error': str(e)}
                    logger.error('Response: {}'.format(response))
                    return self.prepare_response(response, None, status=404, query=request.query)

            status = 200
            logger.info('Response: {}'.format(response))
        except (DeviceException, ServerException, UserException) as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            status = 500
            logger.error('Response: {}'.format(response))

        return self.prepare_response(response, item, status=status, query=request.query, semantic=resource_type.is_semantic())

    async def post(self, request):
        logger.info("POST Request from {}".format(request.url))
        logger.debug('Request: {}'.format(request))
        data = await request.json()
        logger.debug('Body: {}'.format(data))
        uuids = HTTPServer.parse_uuids(request)

        try:
            item = self.root[uuids]
        except KeyError as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            logger.error('Response: {}'.format(response))
            return self.prepare_response(response, None, status=404, query=request.query)

        if isinstance(item, Function):
            try:
                if item.publishes:
                    try:
                        async for item in item.invoke_generator(data["arguments"], legacy_mode=self._legacy_mode):
                            self._publisher.publish('/'.join(uuids), json.dumps(item))
                    except RuntimeError as e:
                        if not isinstance(e.__cause__, StopAsyncIteration):
                            raise e
                    response = {}
                else:
                    response = await item.invoke(data["arguments"], legacy_mode=self._legacy_mode)
                status = 200
                logger.info('Response: {}'.format(response))
            except (DeviceException, ServerException, UserException) as e:
                logger.error(traceback.format_exc())
                response = {'error': str(e)}
                status = 500
                logger.error('Response: {}'.format(response))
        else:
            response, status = {}, 405
            logger.error('Response: {}'.format(response))

        return self.prepare_response(response, item, status=status, query=request.query)

    async def delete(self, request):
        logger.info("DELETE Request from {}".format(request.url))
        logger.debug('Request: {}'.format(request))
        uuids = HTTPServer.parse_uuids(request)

        try:
            item = self.root[uuids[:-1]]
        except KeyError as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            logger.error('Response: {}'.format(response))
            return self.prepare_response(response, None, status=404, query=request.query)

        if not isinstance(item, Component):
            return self.prepare_response({}, None, status=405, query=request.query)
        try:
            await self.loop.run_in_executor(None, functools.partial(item.remove, uuids[-1]))

            if self._scheduler is not None:
                self._scheduler.remove_jobs('/'.join(uuids))
            status = 200
            # logger.info('Response: {}'.format(response))
        except ChildNotFoundException as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            status = 404
            logger.error('Response: {}'.format(response))
        except (DeviceException, ServerException, UserException) as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            status = 500
            logger.error('Response: {}'.format(response))
        return self.prepare_response(response, item, status=status, query=request.query)

    async def options(self, request):
        logger.info("HEAD Request from {}".format(request.url))
        logger.debug('Request: {}'.format(request))
        logger.debug('Query Parameters: {}'.format(request.query_string))
        keys = self._filter_query(request.query)

        try:
            item = self.root[HTTPServer.parse_uuids(request)]
        except KeyError as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            logger.error('Response: {}'.format(response))
            return self.prepare_response(response, None, status=404, query=request.query)

        if not isinstance(item, Variable):
            return self.prepare_response({}, None, status=405, query=request.query)

        response = item.serialize(keys, self._legacy_mode, HTTP_OPTIONS)
        logger.info('Response: {}'.format(response))
        return self.prepare_response(response, item, query=request.query)

    async def patch(self, request):
        logger.info("PATCH Request from {}".format(request.url))
        logger.debug('Request: {}'.format(request))
        data = await request.json()
        logger.debug('Body: {}'.format(data))

        try:
            item = self.root[HTTPServer.parse_uuids(request)]
        except KeyError as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            logger.error('Response: {}'.format(response))
            return self.prepare_response(response, None, status=404, query=request.query)

        if isinstance(item, Parameter):
            try:
                response = await self.loop.run_in_executor(None, item.set, data["value"])
                status = 200
                logger.info('Response: {}'.format(response))
            except ReadOnlyException as e:
                logger.error(traceback.format_exc())
                response = {'error': str(e)}
                status = 403
                logger.error('Response: {}'.format(response))
            except InvokationException as e:
                logger.error(traceback.format_exc())
                response = {'error': str(e)}
                status = 500
                logger.error('Response: {}'.format(response))
        else:
            response, status = {}, 405
            logger.error('Response: {}'.format(response))
        return self.prepare_response(response, item, status=status, query=request.query)

    async def put(self, request):
        logger.info("PUT Request from {}".format(request.url))
        logger.debug('Request: {}'.format(request))
        data = await request.json()
        logger.debug('Body: {}'.format(data))
        uuids = HTTPServer.parse_uuids(request)

        try:
            item = self.root[uuids[:-1]]
        except KeyError as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            logger.error('Response: {}'.format(response))
            return self.prepare_response(response, None, status=404, query=request.query)

        if not isinstance(item, Component):
            return self.prepare_response({}, None, status=405, query=request.query)
        try:
            implementation = await self.loop.run_in_executor(None,
                                                             functools.partial(item.add, uuids[-1], data['class_name'],
                                                                               data['json_file'], *data['args'],
                                                                               **data['kwargs']))
            if self._scheduler is not None:
                self._scheduler.add_jobs(implementation.streams('/'.join(uuids)))
            status = 200
            response = {}
            # logger.info('Response: {}'.format(response))
        except (DeviceException, ServerException, UserException) as e:
            logger.error(traceback.format_exc())
            response = {'error': str(e)}
            status = 500
            logger.error('Response: {}'.format(response))
        return self.prepare_response(response, item, status=status, query=request.query)