diff options
Diffstat (limited to 'src/onapsdk/msb/k8s/definition.py')
-rw-r--r-- | src/onapsdk/msb/k8s/definition.py | 413 |
1 files changed, 4 insertions, 409 deletions
diff --git a/src/onapsdk/msb/k8s/definition.py b/src/onapsdk/msb/k8s/definition.py index 6c0def2..a8c5d6d 100644 --- a/src/onapsdk/msb/k8s/definition.py +++ b/src/onapsdk/msb/k8s/definition.py @@ -12,413 +12,8 @@ # 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 +import warnings +from onapsdk.k8s.definition import Definition, Profile, ConfigurationTemplate # pylint: disable=unused-import -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 +warnings.warn("onapsdk.msb.k8s.definition module is deprecated and will be removed with " + "the next version of ONAP SDK. Use onapsdk.k8s.definition") |