diff options
Diffstat (limited to 'src/onapsdk/msb/k8s/definition.py')
| -rw-r--r-- | src/onapsdk/msb/k8s/definition.py | 424 |
1 files changed, 424 insertions, 0 deletions
diff --git a/src/onapsdk/msb/k8s/definition.py b/src/onapsdk/msb/k8s/definition.py new file mode 100644 index 0000000..6c0def2 --- /dev/null +++ b/src/onapsdk/msb/k8s/definition.py @@ -0,0 +1,424 @@ +"""Definition 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 Iterator +from dataclasses import dataclass + +from onapsdk.utils.jinja import jinja_env +from ..msb_service import MSB + + +# pylint: disable=too-many-arguments, too-few-public-methods +class DefinitionBase(MSB): + """DefinitionBase class.""" + + base_url = f"{MSB.base_url}/api/multicloud-k8s/v1/v1/rb/definition" + + def __init__(self, rb_name: str, + rb_version: str) -> None: + """Definition-Base object initialization. + + Args: + rb_name (str): Definition name + rb_version (str): Definition version + """ + super().__init__() + self.rb_name: str = rb_name + self.rb_version: str = rb_version + + @property + def url(self) -> str: + """URL address for Definition Based calls. + + Returns: + str: URL to RB Definition + + """ + return f"{self.base_url}/{self.rb_name}/{self.rb_version}" + + def delete(self) -> None: + """Delete Definition Based object.""" + self.send_message( + "DELETE", + f"Delete {self.__class__.__name__}", + self.url + ) + + def upload_artifact(self, package: bytes = None): + """Upload artifact. + + Args: + package (bytes): Artifact to be uploaded to multicloud-k8s plugin + + """ + url: str = f"{self.url}/content" + self.send_message( + "POST", + "Upload Artifact content", + url, + data=package, + headers={} + ) + + +class Definition(DefinitionBase): + """Definition class.""" + + def __init__(self, rb_name: str, + rb_version: str, + chart_name: str, + description: str, + labels: dict) -> None: + """Definition object initialization. + + Args: + rb_name (str): Definition name + rb_version (str): Definition version + chart_name (str): Chart name, optional field, will be detected if it is not provided + description (str): Definition description + labels (str): Labels + """ + super().__init__(rb_name, rb_version) + self.rb_name: str = rb_name + self.rb_version: str = rb_version + self.chart_name: str = chart_name + self.description: str = description + self.labels: dict = labels + + @classmethod + def get_all(cls): + """Get all definitions. + + Yields: + Definition: Definition object + + """ + for definition in cls.send_message_json("GET", + "Get definitions", + cls.base_url): + yield cls( + definition["rb-name"], + definition["rb-version"], + definition.get("chart-name"), + definition.get("description"), + definition.get("labels") + ) + + @classmethod + def get_definition_by_name_version(cls, rb_name: str, rb_version: str) -> "Definition": + """Get definition by it's name and version. + + Args: + rb_name (str): definition name + rb_version (str): definition version + + Returns: + Definition: Definition object + + """ + url: str = f"{cls.base_url}/{rb_name}/{rb_version}" + definition: dict = cls.send_message_json( + "GET", + "Get definition", + url + ) + return cls( + definition["rb-name"], + definition["rb-version"], + definition.get("chart-name"), + definition.get("description"), + definition.get("labels") + ) + + @classmethod + def create(cls, rb_name: str, + rb_version: str, + chart_name: str = "", + description: str = "", + labels=None) -> "Definition": + """Create Definition. + + Args: + rb_name (str): Definition name + rb_version (str): Definition version + chart_name (str): Chart name, optional field, will be detected if it is not provided + description (str): Definition description + labels (str): Labels + + Returns: + Definition: Created object + + """ + if labels is None: + labels = {} + url: str = f"{cls.base_url}" + cls.send_message( + "POST", + "Create definition", + url, + data=jinja_env().get_template("multicloud_k8s_add_definition.json.j2").render( + rb_name=rb_name, + rb_version=rb_version, + chart_name=chart_name, + description=description, + labels=labels + ) + ) + return cls.get_definition_by_name_version(rb_name, rb_version) + + def create_profile(self, profile_name: str, + namespace: str, + kubernetes_version: str, + release_name=None) -> "Profile": + """Create Profile for Definition. + + Args: + profile_name (str): Name of profile + namespace (str): Namespace that service is created in + kubernetes_version (str): Required Kubernetes version + release_name (str): Release name + + Returns: + Profile: Created object + + """ + url: str = f"{self.url}/profile" + if release_name is None: + release_name = profile_name + self.send_message( + "POST", + "Create profile for definition", + url, + data=jinja_env().get_template("multicloud_k8s_create_profile_" + "for_definition.json.j2").render( + rb_name=self.rb_name, + rb_version=self.rb_version, + profile_name=profile_name, + release_name=release_name, + namespace=namespace, + kubernetes_version=kubernetes_version + ) + ) + return self.get_profile_by_name(profile_name) + + def get_all_profiles(self) -> Iterator["Profile"]: + """Get all profiles. + + Yields: + Profile: Profile object + + """ + url: str = f"{self.url}/profile" + + for profile in self.send_message_json("GET", + "Get profiles", + url): + yield Profile( + profile["rb-name"], + profile["rb-version"], + profile["profile-name"], + profile["namespace"], + profile.get("kubernetes-version"), + profile.get("labels"), + profile.get("release-name") + ) + + def get_profile_by_name(self, profile_name: str) -> "Profile": + """Get profile by it's name. + + Args: + profile_name (str): profile name + + Returns: + Profile: Profile object + + """ + url: str = f"{self.url}/profile/{profile_name}" + + profile: dict = self.send_message_json( + "GET", + "Get profile", + url + ) + return Profile( + profile["rb-name"], + profile["rb-version"], + profile["profile-name"], + profile["namespace"], + profile.get("kubernetes-version"), + profile.get("labels"), + profile.get("release-name") + ) + + def get_all_configuration_templates(self): + """Get all configuration templates. + + Yields: + ConfigurationTemplate: ConfigurationTemplate object + + """ + url: str = f"{self.url}/config-template" + + for template in self.send_message_json("GET", + "Get configuration templates", + url): + yield ConfigurationTemplate( + self.rb_name, + self.rb_version, + template["template-name"], + template.get("description") + ) + + def create_configuration_template(self, template_name: str, + description="") -> "ConfigurationTemplate": + """Create configuration template. + + Args: + template_name (str): Name of the template + description (str): Description + + Returns: + ConfigurationTemplate: Created object + + """ + url: str = f"{self.url}/config-template" + + self.send_message( + "POST", + "Create configuration template", + url, + data=jinja_env().get_template("multicloud_k8s_create_configuration_" + "template.json.j2").render( + template_name=template_name, + description=description + ) + ) + + return self.get_configuration_template_by_name(template_name) + + def get_configuration_template_by_name(self, template_name: str) -> "ConfigurationTemplate": + """Get configuration template. + + Args: + template_name (str): Name of the template + + Returns: + ConfigurationTemplate: object + + """ + url: str = f"{self.url}/config-template/{template_name}" + + template: dict = self.send_message_json( + "GET", + "Get Configuration template", + url + ) + return ConfigurationTemplate( + self.rb_name, + self.rb_version, + template["template-name"], + template.get("description") + ) + + +class ProfileBase(DefinitionBase): + """ProfileBase class.""" + + def __init__(self, rb_name: str, + rb_version: str, + profile_name: str) -> None: + """Profile-Base object initialization. + + Args: + rb_name (str): Definition name + rb_version (str): Definition version + profile_name (str): Name of profile + """ + super().__init__(rb_name, rb_version) + self.rb_name: str = rb_name + self.rb_version: str = rb_version + self.profile_name: str = profile_name + + @property + def url(self) -> str: + """URL address for Profile calls. + + Returns: + str: URL to RB Profile + + """ + return f"{super().url}/profile/{self.profile_name}" + + +@dataclass +class Profile(ProfileBase): + """Profile class.""" + + def __init__(self, rb_name: str, + rb_version: str, + profile_name: str, + namespace: str, + kubernetes_version: str, + labels=None, + release_name=None) -> None: + """Profile object initialization. + + Args: + rb_name (str): Definition name + rb_version (str): Definition version + profile_name (str): Name of profile + release_name (str): Release name, if release_name is not provided, + namespace (str): Namespace that service is created in + kubernetes_version (str): Required Kubernetes version + labels (dict): Labels + """ + super().__init__(rb_name, rb_version, profile_name) + if release_name is None: + release_name = profile_name + self.release_name: str = release_name + self.namespace: str = namespace + self.kubernetes_version: str = kubernetes_version + self.labels: dict = labels + if self.labels is None: + self.labels = dict() + + +class ConfigurationTemplate(DefinitionBase): + """ConfigurationTemplate class.""" + + @property + def url(self) -> str: + """URL address for ConfigurationTemplate calls. + + Returns: + str: URL to Configuration template in Multicloud-k8s API. + + """ + return f"{super().url}/config-template/{self.template_name}" + + def __init__(self, rb_name: str, + rb_version: str, + template_name: str, + description="") -> None: + """Configuration-Template object initialization. + + Args: + rb_name (str): Definition name + rb_version (str): Definition version + template_name (str): Configuration template name + description (str): Namespace that service is created in + """ + super().__init__(rb_name, rb_version) + self.template_name: str = template_name + self.description: str = description |