aihub-dag/elyra/contents/handlers.py

#
# Copyright 2018-2022 Elyra Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import json
import os

from jupyter_server.base.handlers import APIHandler
from jupyter_server.utils import url_unescape
from tornado import web

from elyra.contents.parser import ContentParser
from elyra.util.http import HttpErrorMixin
from elyra.util.path import get_expanded_path


class ContentHandler(HttpErrorMixin, APIHandler):
    """Handler to expose REST API to parse envs from a File"""

    content_parser: ContentParser = ContentParser()

    @web.authenticated
    async def post(self, path):
        msg_json = dict(title="Operation not supported.")
        self.write(msg_json)
        self.flush()

    @web.authenticated
    async def get(self, path=""):
        path = path or ""
        path = url_unescape(path)

        self.log.debug(f"Parsing file: {path}")

        try:
            absolute_path = self._get_absolute_path(path)
            properties = self.content_parser.parse(absolute_path)

            # TODO: Validation of model
            self.finish(json.dumps(properties))
        except FileNotFoundError as fnfe:
            raise web.HTTPError(404, str(fnfe)) from fnfe
        except IsADirectoryError as iade:
            raise web.HTTPError(400, str(iade)) from iade
        except Exception as e:
            # Parser could not parse the given file, but this does not necessarily indicate an error with the file.
            # Log the issue and return an empty model so that other user processes are not disrupted.
            self.log.debug(f"Could not parse '{path}': {str(e)}")
            empty_properties = {"env_vars": {}, "inputs": [], "outputs": []}
            self.finish(json.dumps(empty_properties))

    def _get_absolute_path(self, path: str) -> str:
        """Returns the absolute path of 'path' in relation to the configured root_dir.

        Note: This version of `get_absolute_path` is used over the version in util/path.py because
        'path' will have a leading `/` and therefore be considered absolute already when, within
        this handler, it should always be a path relative to root_dir.
        """

        root_dir = get_expanded_path(self.settings["server_root_dir"])
        if path[0] == "/":  # if path starts with a slash, use the follow characters so join can be performed
            path = path[1:]
        absolute_path = os.path.normpath(os.path.join(root_dir, path))
        return absolute_path