diff options
Diffstat (limited to 'src/onapsdk/sdc/__init__.py')
| -rw-r--r-- | src/onapsdk/sdc/__init__.py | 486 |
1 files changed, 486 insertions, 0 deletions
diff --git a/src/onapsdk/sdc/__init__.py b/src/onapsdk/sdc/__init__.py new file mode 100644 index 0000000..15280d9 --- /dev/null +++ b/src/onapsdk/sdc/__init__.py @@ -0,0 +1,486 @@ +"""SDC Element module.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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. +from typing import Any, Dict, List, Optional, Union +from operator import attrgetter +from abc import ABC, abstractmethod + +from requests import Response + +from onapsdk.configuration import settings +from onapsdk.exceptions import APIError, RequestError +from onapsdk.onap_service import OnapService +import onapsdk.constants as const +from onapsdk.utils.jinja import jinja_env +from onapsdk.utils.gui import GuiItem, GuiList + +class SDC(OnapService, ABC): + """Mother Class of all SDC elements.""" + + server: str = "SDC" + base_front_url = settings.SDC_FE_URL + base_back_url = settings.SDC_BE_URL + + def __init__(self, name: str = None) -> None: + """Initialize SDC.""" + super().__init__() + self.name: str = name + + def __eq__(self, other: Any) -> bool: + """ + Check equality for SDC and children. + + Args: + other: another object + + Returns: + bool: True if same object, False if not + + """ + if isinstance(other, type(self)): + return self.name == other.name + return False + + @classmethod + @abstractmethod + def _get_all_url(cls) -> str: + """ + Get URL for all elements in SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @classmethod + @abstractmethod + def _get_objects_list(cls, + result: List[Dict[str, Any]]) -> List['SdcResource']: + """ + Import objects created in SDC. + + Args: + result (Dict[str, Any]): the result returned by SDC in a Dict + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @classmethod + @abstractmethod + def _base_url(cls) -> str: + """ + Give back the base url of Sdc. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @classmethod + @abstractmethod + def _base_create_url(cls) -> str: + """ + Give back the base url of Sdc. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @abstractmethod + def _copy_object(self, obj: 'SDC') -> None: + """ + Copy relevant properties from object. + + Args: + obj (Sdc): the object to "copy" + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @classmethod + @abstractmethod + def import_from_sdc(cls, values: Dict[str, Any]) -> 'SDC': + """ + Import Sdc object from SDC. + + Args: + values (Dict[str, Any]): dict to parse returned from SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @staticmethod + def _get_mapped_version(item: "SDC") -> Optional[Union[float, str]]: + """Map Sdc objects version to float. + + Mostly we need to get the newest version of the requested objects. To do + so we use the version property of them. In most cases it's string + formatted float value, but in some cases (like VSP objects) it isn't. + That method checks if given object has "version" attribute and if it's not + a None it tries to map it's value to float. If it's not possible it + returns the alrady existing value. + + Args: + item (SDC): SDC item to map version to float + + Returns: + Optional[Union[float, str]]: Float format version if possible, + string otherwise. If object doesn't have "version" + attribut returns None. + + """ + if hasattr(item, "version") and item.version is not None: + try: + return float(item.version) + except ValueError: + return item.version + else: + return None + + @classmethod + def get_all(cls, **kwargs) -> List['SDC']: + """ + Get the objects list created in SDC. + + Returns: + the list of the objects + + """ + cls._logger.info("retrieving all objects of type %s from SDC", + cls.__name__) + url = cls._get_all_url() + objects = [] + + try: + result = \ + cls.send_message_json('GET', "get {}s".format(cls.__name__), + url, **kwargs) + + for obj_info in cls._get_objects_list(result): + objects.append(cls.import_from_sdc(obj_info)) + + except APIError as exc: + cls._logger.debug("Couldn't get %s: %s", cls.__name__, exc) + except KeyError as exc: + cls._logger.debug("Invalid result dictionary: %s", exc) + + cls._logger.debug("number of %s returned: %s", cls.__name__, + len(objects)) + return objects + + def exists(self) -> bool: + """ + Check if object already exists in SDC and update infos. + + Returns: + True if exists, False either + + """ + self._logger.debug("check if %s %s exists in SDC", + type(self).__name__, self.name) + objects = self.get_all() + + self._logger.debug("filtering objects of all versions to be %s", + self.name) + relevant_objects = list(filter(lambda obj: obj == self, objects)) + + if not relevant_objects: + + self._logger.info("%s %s doesn't exist in SDC", + type(self).__name__, self.name) + return False + + if hasattr(self, 'version_filter') and self.version_filter is not None: # pylint: disable=no-member + + self._logger.debug("filtering %s objects by version %s", + self.name, self.version_filter) # pylint: disable=no-member + + all_versioned = filter( + lambda obj: obj.version == self.version_filter, relevant_objects) # pylint: disable=no-member + + try: + versioned_object = next(all_versioned) + except StopIteration: + self._logger.info("Version %s of %s %s, doesn't exist in SDC", + self.version_filter, type(self).__name__, # pylint: disable=no-member + self.name) + return False + + else: + versioned_object = max(relevant_objects, key=self._get_mapped_version) + + self._logger.info("%s found, updating information", type(self).__name__) + self._copy_object(versioned_object) + return True + + @classmethod + def get_guis(cls) -> GuiItem: + """Retrieve the status of the SDC GUIs. + + Only one GUI is referenced for SDC + the SDC Front End + + Return the list of GUIs + """ + gui_url = settings.SDC_GUI_SERVICE + sdc_gui_response = cls.send_message( + "GET", "Get SDC GUI Status", gui_url) + guilist = GuiList([]) + guilist.add(GuiItem( + gui_url, + sdc_gui_response.status_code)) + return guilist + +class SdcOnboardable(SDC, ABC): + """Base class for onboardable SDC resources (Vendors, Services, ...).""" + + ACTION_TEMPLATE: str + ACTION_METHOD: str + + def __init__(self, name: str = None) -> None: + """Initialize the object.""" + super().__init__(name) + self._identifier: str = None + self._status: str = None + self._version: str = None + + @property + def identifier(self) -> str: + """Return and lazy load the identifier.""" + if not self._identifier: + self.load() + return self._identifier + + @property + def status(self) -> str: + """Return and lazy load the status.""" + if self.created() and not self._status: + self.load() + return self._status + + @property + def version(self) -> str: + """Return and lazy load the version.""" + if self.created() and not self._version: + self.load() + return self._version + + @identifier.setter + def identifier(self, value: str) -> None: + """Set value for identifier.""" + self._identifier = value + + @status.setter + def status(self, status: str) -> None: + """Return and lazy load the status.""" + self._status = status + + @version.setter + def version(self, version: str) -> None: + """Return and lazy load the status.""" + self._version = version + + def created(self) -> bool: + """Determine if SDC is created.""" + if self.name and not self._identifier: + return self.exists() + return bool(self._identifier) + + def submit(self) -> None: + """Submit the SDC object in order to enable it.""" + self._logger.info("attempting to certify/sumbit %s %s in SDC", + type(self).__name__, self.name) + if self.status != const.CERTIFIED and self.created(): + self._really_submit() + elif self.status == const.CERTIFIED: + self._logger.warning("%s %s in SDC is already submitted/certified", + type(self).__name__, self.name) + elif not self.created(): + self._logger.warning("%s %s in SDC is not created", + type(self).__name__, self.name) + + def _create(self, template_name: str, **kwargs) -> None: + """Create the object in SDC if not already existing.""" + self._logger.info("attempting to create %s %s in SDC", + type(self).__name__, self.name) + if not self.exists(): + url = "{}/{}".format(self._base_create_url(), self._sdc_path()) + template = jinja_env().get_template(template_name) + data = template.render(**kwargs) + try: + create_result = self.send_message_json('POST', + "create {}".format( + type(self).__name__), + url, + data=data) + except RequestError as exc: + self._logger.error( + "an error occured during creation of %s %s in SDC", + type(self).__name__, self.name) + raise exc + else: + self._logger.info("%s %s is created in SDC", + type(self).__name__, self.name) + self._status = const.DRAFT + self.identifier = self._get_identifier_from_sdc(create_result) + self._version = self._get_version_from_sdc(create_result) + self.update_informations_from_sdc_creation(create_result) + + else: + self._logger.warning("%s %s is already created in SDC", + type(self).__name__, self.name) + + def _action_to_sdc(self, action: str, action_type: str = None, + **kwargs) -> Response: + """ + Really do an action in the SDC. + + Args: + action (str): the action to perform + action_type (str, optional): the type of action + headers (Dict[str, str], optional): headers to use if any + + Returns: + Response: the response + + """ + subpath = self._generate_action_subpath(action) + url = self._action_url(self._base_create_url(), + subpath, + self._version_path(), + action_type=action_type) + template = jinja_env().get_template(self.ACTION_TEMPLATE) + data = template.render(action=action, const=const) + + return self.send_message(self.ACTION_METHOD, + "{} {}".format(action, + type(self).__name__), + url, + data=data, + **kwargs) + + @abstractmethod + def update_informations_from_sdc(self, details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC. + + Args: + details ([type]): [description] + + """ + @abstractmethod + def update_informations_from_sdc_creation(self, + details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC after creation. + + Args: + details ([type]): the details from SDC + + """ + + @abstractmethod + def load(self) -> None: + """ + Load Object information from SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @abstractmethod + def _get_version_from_sdc(self, sdc_infos: Dict[str, Any]) -> str: + """ + Get version from SDC results. + + Args: + sdc_infos (Dict[str, Any]): the result dict from SDC + + Raises: + NotImplementedError: this is an abstract method. + + """ + @abstractmethod + def _get_identifier_from_sdc(self, sdc_infos: Dict[str, Any]) -> str: + """ + Get identifier from SDC results. + + Args: + sdc_infos (Dict[str, Any]): the result dict from SDC + + Raises: + NotImplementedError: this is an abstract method. + + """ + @abstractmethod + def _generate_action_subpath(self, action: str) -> str: + """ + + Generate subpath part of SDC action url. + + Args: + action (str): the action that will be done + + Raises: + NotImplementedError: this is an abstract method. + + """ + @abstractmethod + def _version_path(self) -> str: + """ + Give the end of the path for a version. + + Raises: + NotImplementedError: this is an abstract method. + + """ + @abstractmethod + def _really_submit(self) -> None: + """Really submit the SDC Vf in order to enable it.""" + @staticmethod + @abstractmethod + def _action_url(base: str, + subpath: str, + version_path: str, + action_type: str = None) -> str: + """ + Generate action URL for SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + @classmethod + @abstractmethod + def _sdc_path(cls) -> None: + """Give back the end of SDC path.""" + + @abstractmethod + def onboard(self) -> None: + """Onboard resource. + + Onboarding is a full stack of actions which needs to be done to + make SDC resource ready to use. It depends on the type of object + but most of them needs to be created and submitted. + """ |