diff options
Diffstat (limited to 'src/onapsdk/sdc/sdc_element.py')
| -rw-r--r-- | src/onapsdk/sdc/sdc_element.py | 227 |
1 files changed, 227 insertions, 0 deletions
diff --git a/src/onapsdk/sdc/sdc_element.py b/src/onapsdk/sdc/sdc_element.py new file mode 100644 index 0000000..df513b7 --- /dev/null +++ b/src/onapsdk/sdc/sdc_element.py @@ -0,0 +1,227 @@ +"""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 abc import ABC, abstractmethod +from operator import itemgetter +from typing import Any, Dict, List, Optional + +from onapsdk.sdc import SdcOnboardable +import onapsdk.constants as const + + +class SdcElement(SdcOnboardable, ABC): + """Mother Class of all SDC elements.""" + + ACTION_TEMPLATE = 'sdc_element_action.json.j2' + ACTION_METHOD = 'PUT' + + def __init__(self, name: str = None) -> None: + """Initialize the object.""" + super().__init__(name=name) + self.human_readable_version: Optional[str] = None + + def _get_item_details(self) -> Dict[str, Any]: + """ + Get item details. + + Returns: + Dict[str, Any]: the description of the item + + """ + if self.created(): + url = "{}/items/{}/versions".format(self._base_url(), + self.identifier) + results: Dict[str, Any] = self.send_message_json('GET', 'get item', url) + if results["listCount"] > 1: + items: List[Dict[str, Any]] = results["results"] + return sorted(items, key=itemgetter("creationTime"), reverse=True)[0] + return results["results"][0] + return {} + + def load(self) -> None: + """Load Object information from SDC.""" + vsp_details = self._get_item_details() + if vsp_details: + self._logger.debug("details found, updating") + self.version = vsp_details['id'] + self.human_readable_version = vsp_details["name"] + self.update_informations_from_sdc(vsp_details) + else: + # exists() method check if exists AND update identifier + self.exists() + + def update_informations_from_sdc(self, details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC. + + Args: + details ([type]): [description] + + """ + 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 + + """ + @classmethod + def _base_url(cls) -> str: + """ + Give back the base url of Sdc. + + Returns: + str: the base url + + """ + return "{}/sdc1/feProxy/onboarding-api/v1.0".format(cls.base_front_url) + + @classmethod + def _base_create_url(cls) -> str: + """ + Give back the base url of Sdc. + + Returns: + str: the base url + + """ + return "{}/sdc1/feProxy/onboarding-api/v1.0".format(cls.base_front_url) + + def _generate_action_subpath(self, action: str) -> str: + """ + + Generate subpath part of SDC action url. + + Args: + action (str): the action that will be done + + Returns: + str: the subpath part + + """ + subpath = self._sdc_path() + if action == const.COMMIT: + subpath = "items" + return subpath + + def _version_path(self) -> str: + """ + Give the end of the path for a version. + + Returns: + str: the end of the path + + """ + return "{}/versions/{}".format(self.identifier, self.version) + + @staticmethod + def _action_url(base: str, + subpath: str, + version_path: str, + action_type: str = None) -> str: + """ + Generate action URL for SDC. + + Args: + base (str): base part of url + subpath (str): subpath of url + version_path (str): version path of the url + action_type (str, optional): the type of action. UNUSED here + + Returns: + str: the URL to use + + """ + return "{}/{}/{}/actions".format(base, subpath, version_path) + + @classmethod + def _get_objects_list(cls, result: List[Dict[str, Any]] + ) -> List[Dict[str, Any]]: + """ + Import objects created in SDC. + + Args: + result (Dict[str, Any]): the result returned by SDC in a Dict + + Return: + List[Dict[str, Any]]: the list of objects + + """ + return result['results'] + + @classmethod + def _get_all_url(cls) -> str: + """ + Get URL for all elements in SDC. + + Returns: + str: the url + + """ + return "{}/{}".format(cls._base_url(), cls._sdc_path()) + + def _copy_object(self, obj: 'SdcElement') -> None: + """ + Copy relevant properties from object. + + Args: + obj (SdcElement): the object to "copy" + + """ + self.identifier = obj.identifier + + 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 + + Returns: + str: the version + + """ + return sdc_infos['version']['id'] + + 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 + + Returns: + str: the identifier + + """ + return sdc_infos['itemId'] + + @classmethod + @abstractmethod + def import_from_sdc(cls, values: Dict[str, Any]) -> 'SdcElement': + """ + Import SdcElement from SDC. + + Args: + values (Dict[str, Any]): dict to parse returned from SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + raise NotImplementedError("SdcElement is an abstract class") |