kytos.core.api_server.APIServer._get_napp_metadata() - Code Metrics - Inspection of "New services to decouple kytos-utils" - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#903)

by Rogerio

created 2019-05-13 17:09 UTC

APIServer._get_napp_metadata() A

↳ Parent: kytos.core.api_server

Complexity

Conditions

Size

Total Lines	19
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
cc	3
eloc	9
nop	4
dl	0
loc	19
ccs	0
cts	0
cp	0
crap	12
rs	9.95
c	0
b	0
f	0

"""Module used to handle a API Server."""
import json
import logging
import os
import shutil
import sys
import warnings
import zipfile
from datetime import datetime
from glob import glob
from http import HTTPStatus
from urllib.error import HTTPError, URLError
from urllib.request import urlopen, urlretrieve

from flask import Flask, jsonify, request, send_file
from flask_cors import CORS
from flask_socketio import SocketIO, join_room, leave_room


class APIServer:
    """Api server used to provide Kytos Controller routes."""

    #: tuple: Default Flask HTTP methods.
    DEFAULT_METHODS = ('GET',)
    _NAPP_PREFIX = "/api/{napp.username}/{napp.name}/"
    _CORE_PREFIX = "/api/kytos/core/"

    def __init__(self, app_name, listen='0.0.0.0', port=8181,
                 napps_manager=None, napps_dir=None):
        """Start a Flask+SocketIO server.

        Require controller to get NApps dir and NAppsManager

        Args:
            app_name(string): String representing a App Name
            listen (string): host name used by api server instance
            port (int): Port number used by api server instance
            controller(kytos.core.controller): A controller instance.
        """
        dirname = os.path.dirname(os.path.abspath(__file__))
        self.nappsManager = napps_manager
        self.napps_dir = napps_dir

        self.flask_dir = os.path.join(dirname, '../web-ui')
        self.log = logging.getLogger('api_server')

        self.listen = listen
        self.port = port

        self.app = Flask(app_name, root_path=self.flask_dir,
                         static_folder="dist", static_url_path="/dist")
        self.server = SocketIO(self.app, async_mode='threading')
        self._enable_websocket_rooms()
        # ENABLE CROSS ORIGIN RESOURCE SHARING
        CORS(self.app)

        # Disable trailing slash
        self.app.url_map.strict_slashes = False

        # Update web-ui if necessary
        self.update_web_ui(force=False)

    def _enable_websocket_rooms(self):
        socket = self.server
        socket.on_event('join', join_room)
        socket.on_event('leave', leave_room)

    def run(self):
        """Run the Flask API Server."""
        try:
            self.server.run(self.app, self.listen, self.port)
        except OSError as exception:
            msg = "Couldn't start API Server: {}".format(exception)
            self.log.critical(msg)
            sys.exit(msg)

    def register_rest_endpoint(self, url, function, methods):
        """Deprecate in favor of @rest decorator."""
        warnings.warn("From now on, use @rest decorator.", DeprecationWarning,
                      stacklevel=2)
        if url.startswith('/'):
            url = url[1:]
        self._start_endpoint(f'/kytos/{url}', function, methods=methods)

    def start_api(self):
        """Start this APIServer instance API.

        Start /api/kytos/core/shutdown/ and status/ endpoints, web UI.
        """
        self.register_core_endpoint('shutdown/', self.shutdown_api)
        self.register_core_endpoint('status/', self.status_api)
        self.register_core_endpoint('web/update/<version>/',
                                    self.update_web_ui,
                                    methods=['POST'])
        self.register_core_endpoint('web/update/',
                                    self.update_web_ui,
                                    methods=['POST'])

        self.register_core_napp_services()

        self._register_web_ui()

    def register_core_endpoint(self, rule, function, **options):
        """Register an endpoint with the URL /api/kytos/core/<rule>.

        Not used by NApps, but controller.
        """
        self._start_endpoint(self._CORE_PREFIX + rule, function, **options)

    def _register_web_ui(self):
        """Register routes to the admin-ui homepage."""
        self.app.add_url_rule('/', self.web_ui.__name__, self.web_ui)
        self.app.add_url_rule('/index.html', self.web_ui.__name__, self.web_ui)
        self.app.add_url_rule('/ui/<username>/<napp_name>/<path:filename>',
                              self.static_web_ui.__name__, self.static_web_ui)
        self.app.add_url_rule('/ui/<path:section_name>',
                              self.get_ui_components.__name__,
                              self.get_ui_components)

    @staticmethod
    def status_api():
        """Display kytos status using the route ``/kytos/status/``."""
        return '{"response": "running"}', HTTPStatus.CREATED.value

    def stop_api_server(self):
        """Send a shutdown request to stop Api Server."""
        try:
            url = f'http://127.0.0.1:{self.port}/api/kytos/core/shutdown'
            urlopen(url)
        except URLError:
            pass

    def shutdown_api(self):
        """Handle shutdown requests received by Api Server.

        This method must be called by kytos using the method
        stop_api_server, otherwise this request will be ignored.
        """
        allowed_host = ['127.0.0.1:'+str(self.port),
                        'localhost:'+str(self.port)]
        if request.host not in allowed_host:
            return "", HTTPStatus.FORBIDDEN.value

        self.server.stop()

        return 'Server shutting down...', HTTPStatus.OK.value

    def static_web_ui(self, username, napp_name, filename):
        """Serve static files from installed napps."""
        path = f"{self.napps_dir}/{username}/{napp_name}/ui/{filename}"
        if os.path.exists(path):
            return send_file(path)
        return "", HTTPStatus.NOT_FOUND.value

    def get_ui_components(self, section_name):
        """Return all napps ui components from an specific section.

        The component name generated will have the following structure:
        {username}-{nappname}-{component-section}-{filename}`

        Args:
            section_name (str): Specific section name

        Returns:
            str: Json with a list of all components found.

        """
        section_name = '*' if section_name == "all" else section_name
        path = f"{self.napps_dir}/*/*/ui/{section_name}/*.kytos"
        components = []
        for name in glob(path):
            dirs_name = name.split('/')
            dirs_name.remove('ui')

            component_name = '-'.join(dirs_name[-4:]).replace('.kytos', '')
            url = f'ui/{"/".join(dirs_name[-4:])}'
            component = {'name': component_name, 'url': url}
            components.append(component)
        return jsonify(components)

    def web_ui(self):
        """Serve the index.html page for the admin-ui."""
        return send_file(f"{self.flask_dir}/index.html")

    def update_web_ui(self, version='latest', force=True):
        """Update the static files for the Web UI.

        Download the latest files from the UI github repository and update them
        in the ui folder.
        The repository link is currently hardcoded here.
        """
        if version == 'latest':
            try:
                url = 'https://api.github.com/repos/kytos/ui/releases/latest'
                response = urlopen(url)
                data = response.readlines()[0]
                version = json.loads(data)['tag_name']
            except URLError:
                version = '1.1.1'

        repository = "https://github.com/kytos/ui"
        uri = repository + f"/releases/download/{version}/latest.zip"

        if not os.path.exists(self.flask_dir) or force:
            # download from github
            try:
                package = urlretrieve(uri)[0]
            except HTTPError:
                return f"Uri not found {uri}."

            # test downloaded zip file
            zip_ref = zipfile.ZipFile(package, 'r')

            if zip_ref.testzip() is not None:
                return f'Zip file from {uri} is corrupted.'

            # backup the old web-ui files and create a new web-ui folder
            if os.path.exists(self.flask_dir):
                date = datetime.now().strftime("%Y%m%d%H%M%S")
                shutil.move(self.flask_dir, f"{self.flask_dir}-{date}")
                os.mkdir(self.flask_dir)

            # unzip and extract files to web-ui/*
            zip_ref.extractall(self.flask_dir)
            zip_ref.close()

        return "updated the web ui"

    # BEGIN decorator methods

    @staticmethod
    def decorate_as_endpoint(rule, **options):
        """Decorate methods as REST endpoints.

        Example for URL ``/api/myusername/mynapp/sayhello/World``:

        .. code-block:: python3

           from flask.json import jsonify
           from kytos.core.napps import rest

           @rest('sayhello/<string:name>')
           def say_hello(name):
               return jsonify({"data": f"Hello, {name}!"})

        ``@rest`` parameters are the same as Flask's ``@app.route``. You can
        also add ``methods=['POST']``, for example.

        As we don't have the NApp instance now, we store the parameters in a
        method attribute in order to add the route later, after we have both
        APIServer and NApp instances.
        """
        def store_route_params(function):
            """Store ``Flask`` ``@route`` parameters in a method attribute.

            There can be many @route decorators in a single function.
            """
            # To support any order: @classmethod, @rest or @rest, @classmethod
            # class and static decorators return a descriptor with the function
            # in __func__.
            if isinstance(function, (classmethod, staticmethod)):
                inner = function.__func__
            else:
                inner = function
            # Add route parameters
            if not hasattr(inner, 'route_params'):
                inner.route_params = []
            inner.route_params.append((rule, options))
            # Return the same function, now with "route_params" attribute
            return function
        return store_route_params

    def register_napp_endpoints(self, napp):
        """Add all NApp REST endpoints with @rest decorator.

        URLs will be prefixed with ``/api/{username}/{napp_name}/``.

        Args:
            napp (Napp): Napp instance to register new endpoints.
        """
        for function in self._get_decorated_functions(napp):
            for rule, options in function.route_params:
                absolute_rule = self.get_absolute_rule(rule, napp)
                self._start_endpoint(absolute_rule, function, **options)

    @staticmethod
    def _get_decorated_functions(napp):
        """Return ``napp``'s methods having the @rest decorator."""
        for name in dir(napp):
            if not name.startswith('_'):  # discarding private names
                pub_attr = getattr(napp, name)
                if callable(pub_attr) and hasattr(pub_attr, 'route_params'):
                    yield pub_attr

    @classmethod
    def get_absolute_rule(cls, rule, napp):
        """Prefix the rule, e.g. "flow" to "/api/user/napp/flow".

        This code is used by kytos-utils when generating an OpenAPI skel.
        """
        # Flask does require 2 slashes if specified, so we remove a starting
        # slash if applicable.
        relative_rule = rule[1:] if rule.startswith('/') else rule
        return cls._NAPP_PREFIX.format(napp=napp) + relative_rule

    # END decorator methods

    def _start_endpoint(self, rule, function, **options):
        """Start ``function``'s endpoint.

        Forward parameters to ``Flask.add_url_rule`` mimicking Flask
        ``@route`` decorator.
        """
        endpoint = options.pop('endpoint', None)
        self.app.add_url_rule(rule, endpoint, function, **options)
        self.log.info('Started %s - %s', rule,
                      ', '.join(options.get('methods', self.DEFAULT_METHODS)))

    def remove_napp_endpoints(self, napp):
        """Remove all decorated endpoints.

        Args:
            napp (Napp): Napp instance to look for rest-decorated methods.
        """
        prefix = self._NAPP_PREFIX.format(napp=napp)
        indexes = []
        endpoints = set()
        for index, rule in enumerate(self.app.url_map.iter_rules()):
            if rule.rule.startswith(prefix):
                endpoints.add(rule.endpoint)
                indexes.append(index)
                self.log.info('Stopped %s - %s', rule, ','.join(rule.methods))

        for endpoint in endpoints:
            self.app.view_functions.pop(endpoint)

        for index in reversed(indexes):
            # pylint: disable=protected-access
            self.app.url_map._rules.pop(index)
            # pylint: enable=protected-access

        self.log.info('The Rest endpoints from %s were disabled.', prefix)

    def register_core_napp_services(self):
        """
        Register /kytos/core/ services over NApps.

        It registers enable, disable, install, uninstall NApps that will
        be used by kytos-utils.
        """
        self.register_core_endpoint("napps/<username>/<napp_name>/enable",
                                    self._enable_napp)
        self.register_core_endpoint("napps/<username>/<napp_name>/disable",
                                    self._disable_napp)
        self.register_core_endpoint("napps/<username>/<napp_name>/install",
                                    self._install_napp)
        self.register_core_endpoint("napps/<username>/<napp_name>/uninstall",
                                    self._uninstall_napp)
        self.register_core_endpoint("napps_enabled",
                                    self._list_enabled_napps)
        self.register_core_endpoint("napps_installed",
                                    self._list_installed_napps)
        self.register_core_endpoint(
            "napps/<username>/<napp_name>/metadata/<key>",
            self._get_napp_metadata)

    def _enable_napp(self, username, napp_name):
        """
        Enable an installed NApp.

        :param username: NApps user name
        :param napp_name: NApp name
        :return: JSON content and return code
        """
        # Check if the NApp is installed
        if not self.nappsManager.is_installed(username, napp_name):
            return '{"response": "not installed"}', \
                   HTTPStatus.BAD_REQUEST.value

        # Check if the NApp is already been enabled
        if not self.nappsManager.is_enabled(username, napp_name):
            self.nappsManager.enable(username, napp_name)

        # Check if NApp is enabled
        if not self.nappsManager.is_enabled(username, napp_name):
            # If it is not enabled an admin user must check the log file
            return '{"response": "error"}', \
                   HTTPStatus.INTERNAL_SERVER_ERROR.value

        return '{"response": "enabled"}', HTTPStatus.OK.value

    def _disable_napp(self, username, napp_name):
        """
        Disable an installed NApp.

        :param username: NApps user name
        :param napp_name: NApp name
        :return: JSON content and return code
        """
        # Check if the NApp is installed
        if not self.nappsManager.is_installed(username, napp_name):
            return '{"response": "not installed"}', \
                   HTTPStatus.BAD_REQUEST.value

        # Check if the NApp is enabled
        if self.nappsManager.is_enabled(username, napp_name):
            self.nappsManager.disable(username, napp_name)

        # Check if NApp is still enabled
        if self.nappsManager.is_enabled(username, napp_name):
            # If it is still enabled an admin user must check the log file
            return '{"response": "error"}', \
                   HTTPStatus.INTERNAL_SERVER_ERROR.value

        return '{"response": "disabled"}', \
               HTTPStatus.OK.value

    def _install_napp(self, username, napp_name):
        # Check if the NApp is installed
        if self.nappsManager.is_installed(username, napp_name):
            return '{"response": "installed"}', HTTPStatus.OK.value

        napp = "{}/{}".format(username, napp_name)

        # Try to install the napp
        if not self.nappsManager.install(napp, enable=False):
            # If it is not installed an admin user must check the log file
            return '{"response": "error"}', \
                   HTTPStatus.INTERNAL_SERVER_ERROR.value

        return '{"response": "installed"}', HTTPStatus.OK.value

    def _uninstall_napp(self, username, napp_name):
        # Check if the NApp is installed
        if self.nappsManager.is_installed(username, napp_name):
            # Try to unload/uninstall the napp
            if not self.nappsManager.uninstall(username, napp_name):
                # If it is not uninstalled admin user must check the log file
                return '{"response": "error"}', \
                       HTTPStatus.INTERNAL_SERVER_ERROR.value

        return '{"response": "uninstalled"}', HTTPStatus.OK.value

    def _list_enabled_napps(self):
        """Sorted list of (username, napp_name) of enabled napps."""
        serialized_dict = json.dumps(
                            self.nappsManager.get_enabled_napps(),
                            default=lambda a: [a.username, a.name])

        return '{"napps": %s}' % serialized_dict, HTTPStatus.OK.value

    def _list_installed_napps(self):
        """Sorted list of (username, napp_name) of installed napps."""
        serialized_dict = json.dumps(
                            self.nappsManager.get_installed_napps(),
                            default=lambda a: [a.username, a.name])

        return '{"napps": %s}' % serialized_dict, HTTPStatus.OK.value

    def _get_napp_metadata(self, username, napp_name, key):
        """Get NApp metadata value.

        For safety reasons, only some keys can be retrieved:
        napp_dependencies, description, version.

        """
        _VALID_KEYS = ['napp_dependencies', 'description', 'version']

        if not self.nappsManager.is_installed(username, napp_name):
            return "Napp is not installed.", HTTPStatus.BAD_REQUEST.value

        if key not in _VALID_KEYS:
            return "Invalid key.", HTTPStatus.BAD_REQUEST.value

        data = self.nappsManager.get_napp_metadata(username, napp_name, key)
        serialized_dict = json.dumps({key: data})

        return '%s' % serialized_dict, HTTPStatus.OK.value


1		"""Module used to handle a API Server."""
2	1	import json
3	1	import logging
4	1	import os
5	1	import shutil
6	1	import sys
7	1	import warnings
8	1	import zipfile
9	1	from datetime import datetime
10	1	from glob import glob
11	1	from http import HTTPStatus
12	1	from urllib.error import HTTPError, URLError
13		from urllib.request import urlopen, urlretrieve
14	1
15	1	from flask import Flask, jsonify, request, send_file
16	1	from flask_cors import CORS
17		from flask_socketio import SocketIO, join_room, leave_room
18
19	1
20		class APIServer:
21		"""Api server used to provide Kytos Controller routes."""
22
23	1	#: tuple: Default Flask HTTP methods.
24	1	DEFAULT_METHODS = ('GET',)
25	1	_NAPP_PREFIX = "/api/{napp.username}/{napp.name}/"
26		_CORE_PREFIX = "/api/kytos/core/"
27	1
28		def __init__(self, app_name, listen='0.0.0.0', port=8181,
29		napps_manager=None, napps_dir=None):
30		"""Start a Flask+SocketIO server.
31
32		Require controller to get NApps dir and NAppsManager
33
34		Args:
35		app_name(string): String representing a App Name
36		listen (string): host name used by api server instance
37	1	port (int): Port number used by api server instance
38	1	controller(kytos.core.controller): A controller instance.
39	1	"""
40		dirname = os.path.dirname(os.path.abspath(__file__))
41	1	self.nappsManager = napps_manager
42	1	self.napps_dir = napps_dir
43
44	1	self.flask_dir = os.path.join(dirname, '../web-ui')
45		self.log = logging.getLogger('api_server')
46	1
47	1	self.listen = listen
48		self.port = port
49	1
50		self.app = Flask(app_name, root_path=self.flask_dir,
51		static_folder="dist", static_url_path="/dist")
52	1	self.server = SocketIO(self.app, async_mode='threading')
53		self._enable_websocket_rooms()
54		# ENABLE CROSS ORIGIN RESOURCE SHARING
55	1	CORS(self.app)
56	1
57		# Disable trailing slash
58	1	self.app.url_map.strict_slashes = False
59	1
60	1	# Update web-ui if necessary
61	1	self.update_web_ui(force=False)
62
63	1	def _enable_websocket_rooms(self):
64		socket = self.server
65		socket.on_event('join', join_room)
66		socket.on_event('leave', leave_room)
67
68		def run(self):
69		"""Run the Flask API Server."""
70		try:
71		self.server.run(self.app, self.listen, self.port)
72	1	except OSError as exception:
73		msg = "Couldn't start API Server: {}".format(exception)
74	1	self.log.critical(msg)
75		sys.exit(msg)
76	1
77		def register_rest_endpoint(self, url, function, methods):
78	1	"""Deprecate in favor of @rest decorator."""
79		warnings.warn("From now on, use @rest decorator.", DeprecationWarning,
80	1	stacklevel=2)
81		if url.startswith('/'):
82		url = url[1:]
83		self._start_endpoint(f'/kytos/{url}', function, methods=methods)
84
85	1	def start_api(self):
86	1	"""Start this APIServer instance API.
87	1
88		Start /api/kytos/core/shutdown/ and status/ endpoints, web UI.
89		"""
90	1	self.register_core_endpoint('shutdown/', self.shutdown_api)
91		self.register_core_endpoint('status/', self.status_api)
92		self.register_core_endpoint('web/update/<version>/',
93	1	self.update_web_ui,
94		methods=['POST'])
95	1	self.register_core_endpoint('web/update/',
96		self.update_web_ui,
97		methods=['POST'])
98
99		self.register_core_napp_services()
100	1
101		self._register_web_ui()
102	1
103		def register_core_endpoint(self, rule, function, **options):
104	1	"""Register an endpoint with the URL /api/kytos/core/<rule>.
105	1
106	1	Not used by NApps, but controller.
107		"""
108	1	self._start_endpoint(self._CORE_PREFIX + rule, function, **options)
109
110		def _register_web_ui(self):
111		"""Register routes to the admin-ui homepage."""
112	1	self.app.add_url_rule('/', self.web_ui.__name__, self.web_ui)
113		self.app.add_url_rule('/index.html', self.web_ui.__name__, self.web_ui)
114		self.app.add_url_rule('/ui/<username>/<napp_name>/<path:filename>',
115		self.static_web_ui.__name__, self.static_web_ui)
116		self.app.add_url_rule('/ui/<path:section_name>',
117	1	self.get_ui_components.__name__,
118		self.get_ui_components)
119
120		@staticmethod
121		def status_api():
122		"""Display kytos status using the route ``/kytos/status/``."""
123		return '{"response": "running"}', HTTPStatus.CREATED.value
124
125	1	def stop_api_server(self):
126		"""Send a shutdown request to stop Api Server."""
127		try:
128		url = f'http://127.0.0.1:{self.port}/api/kytos/core/shutdown'
129		urlopen(url)
130		except URLError:
131		pass
132
133		def shutdown_api(self):
134		"""Handle shutdown requests received by Api Server.
135
136		This method must be called by kytos using the method
137		stop_api_server, otherwise this request will be ignored.
138		"""
139		allowed_host = ['127.0.0.1:'+str(self.port),
140	1	'localhost:'+str(self.port)]
141		if request.host not in allowed_host:
142		return "", HTTPStatus.FORBIDDEN.value
143
144		self.server.stop()
145
146		return 'Server shutting down...', HTTPStatus.OK.value
147	1
148		def static_web_ui(self, username, napp_name, filename):
149		"""Serve static files from installed napps."""
150		path = f"{self.napps_dir}/{username}/{napp_name}/ui/{filename}"
151		if os.path.exists(path):
152		return send_file(path)
153		return "", HTTPStatus.NOT_FOUND.value
154
155		def get_ui_components(self, section_name):
156		"""Return all napps ui components from an specific section.
157
158		The component name generated will have the following structure:
159		{username}-{nappname}-{component-section}-{filename}`
160
161		Args:
162		section_name (str): Specific section name
163
164		Returns:
165		str: Json with a list of all components found.
166
167		"""
168		section_name = '*' if section_name == "all" else section_name
169		path = f"{self.napps_dir}///ui/{section_name}/*.kytos"
170		components = []
171		for name in glob(path):
172		dirs_name = name.split('/')
173	1	dirs_name.remove('ui')
174
175		component_name = '-'.join(dirs_name[-4:]).replace('.kytos', '')
176		url = f'ui/{"/".join(dirs_name[-4:])}'
177	1	component = {'name': component_name, 'url': url}
178		components.append(component)
179		return jsonify(components)
180
181		def web_ui(self):
182		"""Serve the index.html page for the admin-ui."""
183		return send_file(f"{self.flask_dir}/index.html")
184	1
185	1	def update_web_ui(self, version='latest', force=True):
186	1	"""Update the static files for the Web UI.
187	1
188	1	Download the latest files from the UI github repository and update them
189	1	in the ui folder.
190		The repository link is currently hardcoded here.
191		"""
192		if version == 'latest':
193	1	try:
194	1	url = 'https://api.github.com/repos/kytos/ui/releases/latest'
195		response = urlopen(url)
196	1	data = response.readlines()[0]
197		version = json.loads(data)['tag_name']
198	1	except URLError:
199	1	version = '1.1.1'
200
201		repository = "https://github.com/kytos/ui"
202		uri = repository + f"/releases/download/{version}/latest.zip"
203
204	1	if not os.path.exists(self.flask_dir) or force:
205		# download from github
206	1	try:
207		package = urlretrieve(uri)[0]
208		except HTTPError:
209		return f"Uri not found {uri}."
210	1
211		# test downloaded zip file
212		zip_ref = zipfile.ZipFile(package, 'r')
213
214		if zip_ref.testzip() is not None:
215		return f'Zip file from {uri} is corrupted.'
216	1
217	1	# backup the old web-ui files and create a new web-ui folder
218		if os.path.exists(self.flask_dir):
219	1	date = datetime.now().strftime("%Y%m%d%H%M%S")
220		shutil.move(self.flask_dir, f"{self.flask_dir}-{date}")
221		os.mkdir(self.flask_dir)
222
223	1	# unzip and extract files to web-ui/*
224		zip_ref.extractall(self.flask_dir)
225		zip_ref.close()
226
227		return "updated the web ui"
228
229		# BEGIN decorator methods
230
231		@staticmethod
232		def decorate_as_endpoint(rule, **options):
233		"""Decorate methods as REST endpoints.
234
235		Example for URL ``/api/myusername/mynapp/sayhello/World``:
236
237		.. code-block:: python3
238
239		from flask.json import jsonify
240		from kytos.core.napps import rest
241
242		@rest('sayhello/<string:name>')
243		def say_hello(name):
244		return jsonify({"data": f"Hello, {name}!"})
245	1
246		``@rest`` parameters are the same as Flask's ``@app.route``. You can
247		also add ``methods=['POST']``, for example.
248
249		As we don't have the NApp instance now, we store the parameters in a
250		method attribute in order to add the route later, after we have both
251		APIServer and NApp instances.
252		"""
253	1	def store_route_params(function):
254	1	"""Store ``Flask`` ``@route`` parameters in a method attribute.
255
256	1	There can be many @route decorators in a single function.
257		"""
258	1	# To support any order: @classmethod, @rest or @rest, @classmethod
259	1	# class and static decorators return a descriptor with the function
260	1	# in __func__.
261		if isinstance(function, (classmethod, staticmethod)):
262	1	inner = function.__func__
263	1	else:
264		inner = function
265	1	# Add route parameters
266		if not hasattr(inner, 'route_params'):
267		inner.route_params = []
268		inner.route_params.append((rule, options))
269		# Return the same function, now with "route_params" attribute
270		return function
271		return store_route_params
272
273	1	def register_napp_endpoints(self, napp):
274	1	"""Add all NApp REST endpoints with @rest decorator.
275	1
276	1	URLs will be prefixed with ``/api/{username}/{napp_name}/``.
277
278	1	Args:
279		napp (Napp): Napp instance to register new endpoints.
280		"""
281	1	for function in self._get_decorated_functions(napp):
282	1	for rule, options in function.route_params:
283	1	absolute_rule = self.get_absolute_rule(rule, napp)
284	1	self._start_endpoint(absolute_rule, function, **options)
285	1
286		@staticmethod
287	1	def _get_decorated_functions(napp):
288		"""Return ``napp``'s methods having the @rest decorator."""
289		for name in dir(napp):
290		if not name.startswith('_'): # discarding private names
291		pub_attr = getattr(napp, name)
292		if callable(pub_attr) and hasattr(pub_attr, 'route_params'):
293		yield pub_attr
294
295	1	@classmethod
296	1	def get_absolute_rule(cls, rule, napp):
297		"""Prefix the rule, e.g. "flow" to "/api/user/napp/flow".
298
299		This code is used by kytos-utils when generating an OpenAPI skel.
300	1	"""
301		# Flask does require 2 slashes if specified, so we remove a starting
302		# slash if applicable.
303		relative_rule = rule[1:] if rule.startswith('/') else rule
304		return cls._NAPP_PREFIX.format(napp=napp) + relative_rule
305
306	1	# END decorator methods
307	1
308	1	def _start_endpoint(self, rule, function, **options):
309		"""Start ``function``'s endpoint.
310
311	1	Forward parameters to ``Flask.add_url_rule`` mimicking Flask
312		``@route`` decorator.
313		"""
314		endpoint = options.pop('endpoint', None)
315		self.app.add_url_rule(rule, endpoint, function, **options)
316		self.log.info('Started %s - %s', rule,
317	1	', '.join(options.get('methods', self.DEFAULT_METHODS)))
318	1
319	1	def remove_napp_endpoints(self, napp):
320	1	"""Remove all decorated endpoints.
321	1
322	1	Args:
323	1	napp (Napp): Napp instance to look for rest-decorated methods.
324	1	"""
325		prefix = self._NAPP_PREFIX.format(napp=napp)
326	1	indexes = []
327	1	endpoints = set()
328		for index, rule in enumerate(self.app.url_map.iter_rules()):
329	1	if rule.rule.startswith(prefix):
330		endpoints.add(rule.endpoint)
331	1	indexes.append(index)
332		self.log.info('Stopped %s - %s', rule, ','.join(rule.methods))
333
334	1	for endpoint in endpoints:
335		self.app.view_functions.pop(endpoint)
336
337		for index in reversed(indexes):
338		# pylint: disable=protected-access
339		self.app.url_map._rules.pop(index)
340		# pylint: enable=protected-access
341
342		self.log.info('The Rest endpoints from %s were disabled.', prefix)
343
344		def register_core_napp_services(self):
345		"""
346		Register /kytos/core/ services over NApps.
347
348		It registers enable, disable, install, uninstall NApps that will
349		be used by kytos-utils.
350		"""
351		self.register_core_endpoint("napps/<username>/<napp_name>/enable",
352		self._enable_napp)
353		self.register_core_endpoint("napps/<username>/<napp_name>/disable",
354		self._disable_napp)
355		self.register_core_endpoint("napps/<username>/<napp_name>/install",
356		self._install_napp)
357		self.register_core_endpoint("napps/<username>/<napp_name>/uninstall",
358		self._uninstall_napp)
359		self.register_core_endpoint("napps_enabled",
360		self._list_enabled_napps)
361		self.register_core_endpoint("napps_installed",
362		self._list_installed_napps)
363		self.register_core_endpoint(
364		"napps/<username>/<napp_name>/metadata/<key>",
365		self._get_napp_metadata)
366
367		def _enable_napp(self, username, napp_name):
368		"""
369		Enable an installed NApp.
370
371		:param username: NApps user name
372		:param napp_name: NApp name
373		:return: JSON content and return code
374		"""
375		# Check if the NApp is installed
376		if not self.nappsManager.is_installed(username, napp_name):
377		return '{"response": "not installed"}', \
378		HTTPStatus.BAD_REQUEST.value
379
380		# Check if the NApp is already been enabled
381		if not self.nappsManager.is_enabled(username, napp_name):
382		self.nappsManager.enable(username, napp_name)
383
384		# Check if NApp is enabled
385		if not self.nappsManager.is_enabled(username, napp_name):
386		# If it is not enabled an admin user must check the log file
387		return '{"response": "error"}', \
388		HTTPStatus.INTERNAL_SERVER_ERROR.value
389
390		return '{"response": "enabled"}', HTTPStatus.OK.value
391
392		def _disable_napp(self, username, napp_name):
393		"""
394		Disable an installed NApp.
395
396		:param username: NApps user name
397		:param napp_name: NApp name
398		:return: JSON content and return code
399		"""
400		# Check if the NApp is installed
401		if not self.nappsManager.is_installed(username, napp_name):
402		return '{"response": "not installed"}', \
403		HTTPStatus.BAD_REQUEST.value
404
405		# Check if the NApp is enabled
406		if self.nappsManager.is_enabled(username, napp_name):
407		self.nappsManager.disable(username, napp_name)
408
409		# Check if NApp is still enabled
410		if self.nappsManager.is_enabled(username, napp_name):
411		# If it is still enabled an admin user must check the log file
412		return '{"response": "error"}', \
413		HTTPStatus.INTERNAL_SERVER_ERROR.value
414
415		return '{"response": "disabled"}', \
416		HTTPStatus.OK.value
417
418		def _install_napp(self, username, napp_name):
419		# Check if the NApp is installed
420		if self.nappsManager.is_installed(username, napp_name):
421		return '{"response": "installed"}', HTTPStatus.OK.value
422
423		napp = "{}/{}".format(username, napp_name)
424
425		# Try to install the napp
426		if not self.nappsManager.install(napp, enable=False):
427		# If it is not installed an admin user must check the log file
428		return '{"response": "error"}', \
429		HTTPStatus.INTERNAL_SERVER_ERROR.value
430
431		return '{"response": "installed"}', HTTPStatus.OK.value
432
433		def _uninstall_napp(self, username, napp_name):
434		# Check if the NApp is installed
435		if self.nappsManager.is_installed(username, napp_name):
436		# Try to unload/uninstall the napp
437		if not self.nappsManager.uninstall(username, napp_name):
438		# If it is not uninstalled admin user must check the log file
439		return '{"response": "error"}', \
440		HTTPStatus.INTERNAL_SERVER_ERROR.value
441
442		return '{"response": "uninstalled"}', HTTPStatus.OK.value
443
444		def _list_enabled_napps(self):
445		"""Sorted list of (username, napp_name) of enabled napps."""
446		serialized_dict = json.dumps(
447		self.nappsManager.get_enabled_napps(),
448		default=lambda a: [a.username, a.name])
449
450		return '{"napps": %s}' % serialized_dict, HTTPStatus.OK.value
451
452		def _list_installed_napps(self):
453		"""Sorted list of (username, napp_name) of installed napps."""
454		serialized_dict = json.dumps(
455		self.nappsManager.get_installed_napps(),
456		default=lambda a: [a.username, a.name])
457
458		return '{"napps": %s}' % serialized_dict, HTTPStatus.OK.value
459
460		def _get_napp_metadata(self, username, napp_name, key):
461		"""Get NApp metadata value.
462
463		For safety reasons, only some keys can be retrieved:
464		napp_dependencies, description, version.
465
466		"""
467		_VALID_KEYS = ['napp_dependencies', 'description', 'version']
468
469		if not self.nappsManager.is_installed(username, napp_name):
470		return "Napp is not installed.", HTTPStatus.BAD_REQUEST.value
471
472		if key not in _VALID_KEYS:
473		return "Invalid key.", HTTPStatus.BAD_REQUEST.value
474
475		data = self.nappsManager.get_napp_metadata(username, napp_name, key)
476		serialized_dict = json.dumps({key: data})
477
478		return '%s' % serialized_dict, HTTPStatus.OK.value
479

kytos / kytos

Pull Request — master (#903)

APIServer._get_napp_metadata() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like