diff options
Diffstat (limited to 'src/onapsdk/cds/blueprint_model.py')
| -rw-r--r-- | src/onapsdk/cds/blueprint_model.py | 222 |
1 files changed, 222 insertions, 0 deletions
diff --git a/src/onapsdk/cds/blueprint_model.py b/src/onapsdk/cds/blueprint_model.py new file mode 100644 index 0000000..7976001 --- /dev/null +++ b/src/onapsdk/cds/blueprint_model.py @@ -0,0 +1,222 @@ +"""CDS Blueprint Models 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 onapsdk.exceptions import ResourceNotFound # for custom exceptions + +from .blueprint import Blueprint +from .cds_element import CdsElement + + +class BlueprintModel(CdsElement): # pylint: disable=too-many-instance-attributes + """Blueprint Model class. + + Represents blueprint models in CDS + """ + + def __init__(self, # pylint: disable=too-many-arguments + blueprint_model_id: str, + artifact_uuid: str = None, + artifact_type: str = None, + artifact_version: str = None, + artifact_description: str = None, + internal_version: str = None, + created_date: str = None, + artifact_name: str = None, + published: str = 'N', + updated_by: str = None, + tags: str = None): + """Blueprint Model initialization. + + Args: + blueprint_model_id (str): Blueprint model identifier + artifact_uuid (str): Blueprint model uuid + artifact_type (str): Blueprint artifact type + artifact_version (str): Blueprint model version + artifact_description (str): Blueprint model description + internal_version (str): Blueprint model internal version + created_date (str): Blueprint model created date + artifact_name (str): Blueprint model name + published (str): Blueprint model publish status - 'N' or 'Y' + updated_by (str): Blueprint model author + tags (str): Blueprint model tags + + """ + super().__init__() + self.blueprint_model_id = blueprint_model_id + self.artifact_uuid = artifact_uuid + self.artifact_type = artifact_type + self.artifact_version = artifact_version + self.artifact_description = artifact_description + self.internal_version = internal_version + self.created_date = created_date + self.artifact_name = artifact_name + self.published = published + self.updated_by = updated_by + self.tags = tags + + def __repr__(self) -> str: + """Representation of object. + + Returns: + str: Object's string representation + + """ + return (f"BlueprintModel(artifact_name='{self.artifact_name}', " + f"blueprint_model_id='{self.blueprint_model_id}')") + + @classmethod + def get_by_id(cls, blueprint_model_id: str) -> "BlueprintModel": + """Retrieve blueprint model with provided ID. + + Args: blueprint_model_id (str): + + Returns: + BlueprintModel: Blueprint model object + + Raises: + ResourceNotFound: Blueprint model with provided ID doesn't exist + + """ + try: + blueprint_model = cls.send_message_json( + "GET", + "Retrieve blueprint", + f"{cls._url}/api/v1/blueprint-model/{blueprint_model_id}", + auth=cls.auth) + + return cls( + blueprint_model_id=blueprint_model["blueprintModel"]['id'], + artifact_uuid=blueprint_model["blueprintModel"]['artifactUUId'], + artifact_type=blueprint_model["blueprintModel"]['artifactType'], + artifact_version=blueprint_model["blueprintModel"]['artifactVersion'], + internal_version=blueprint_model["blueprintModel"]['internalVersion'], + created_date=blueprint_model["blueprintModel"]['createdDate'], + artifact_name=blueprint_model["blueprintModel"]['artifactName'], + published=blueprint_model["blueprintModel"]['published'], + updated_by=blueprint_model["blueprintModel"]['updatedBy'], + tags=blueprint_model["blueprintModel"]['tags'] + ) + + except ResourceNotFound: + raise ResourceNotFound(f"BlueprintModel blueprint_model_id='{blueprint_model_id}" + f" not found") + + @classmethod + def get_by_name_and_version(cls, blueprint_name: str, + blueprint_version: str) -> "BlueprintModel": + """Retrieve blueprint model with provided name and version. + + Args: + blueprint_name (str): Blueprint model name + blueprint_version (str): Blueprint model version + + Returns: + BlueprintModel: Blueprint model object + + Raises: + ResourceNotFound: Blueprint model with provided name and version doesn't exist + + """ + try: + blueprint_model = cls.send_message_json( + "GET", + "Retrieve blueprint", + f"{cls._url}/api/v1/blueprint-model/by-name/{blueprint_name}" + f"/version/{blueprint_version}", + auth=cls.auth) + + return cls( + blueprint_model_id=blueprint_model["blueprintModel"]['id'], + artifact_uuid=blueprint_model["blueprintModel"]['artifactUUId'], + artifact_type=blueprint_model["blueprintModel"]['artifactType'], + artifact_version=blueprint_model["blueprintModel"]['artifactVersion'], + internal_version=blueprint_model["blueprintModel"]['internalVersion'], + created_date=blueprint_model["blueprintModel"]['createdDate'], + artifact_name=blueprint_model["blueprintModel"]['artifactName'], + published=blueprint_model["blueprintModel"]['published'], + updated_by=blueprint_model["blueprintModel"]['updatedBy'], + tags=blueprint_model["blueprintModel"]['tags'] + ) + + except ResourceNotFound: + raise ResourceNotFound(f"BlueprintModel blueprint_name='{blueprint_name}" + f" and blueprint_version='{blueprint_version}' not found") + + @classmethod + def get_all(cls) -> Iterator["BlueprintModel"]: + """Get all blueprint models. + + Yields: + BlueprintModel: BlueprintModel object. + + """ + for blueprint_model in cls.send_message_json( + "GET", + "Retrieve all blueprints", + f"{cls._url}/api/v1/blueprint-model", + auth=cls.auth): + + yield cls( + blueprint_model_id=blueprint_model["blueprintModel"]['id'], + artifact_uuid=blueprint_model["blueprintModel"]['artifactUUId'], + artifact_type=blueprint_model["blueprintModel"]['artifactType'], + artifact_version=blueprint_model["blueprintModel"]['artifactVersion'], + internal_version=blueprint_model["blueprintModel"]['internalVersion'], + created_date=blueprint_model["blueprintModel"]['createdDate'], + artifact_name=blueprint_model["blueprintModel"]['artifactName'], + published=blueprint_model["blueprintModel"]['published'], + updated_by=blueprint_model["blueprintModel"]['updatedBy'], + tags=blueprint_model["blueprintModel"]['tags'] + ) + + def get_blueprint(self) -> Blueprint: + """Get Blueprint object for selected blueprint model. + + Returns: + Blueprint: Blueprint object + + """ + cba_package = self.send_message( + "GET", + "Retrieve selected blueprint object", + f"{self._url}/api/v1/blueprint-model/download/{self.blueprint_model_id}", + auth=self.auth) + + return Blueprint(cba_file_bytes=cba_package.content) + + def save(self, dst_file_path: str): + """Save blueprint model to file. + + Args: + dst_file_path (str): Path of file where blueprint is going to be saved + """ + cba_package = self.send_message( + "GET", + "Retrieve and save selected blueprint", + f"{self._url}/api/v1/blueprint-model/download/{self.blueprint_model_id}", + auth=self.auth) + + with open(dst_file_path, "wb") as content: + for chunk in cba_package.iter_content(chunk_size=128): + content.write(chunk) + + def delete(self): + """Delete blueprint model.""" + self.send_message( + "DELETE", + "Delete blueprint", + f"{self._url}/api/v1/blueprint-model/{self.blueprint_model_id}", + auth=self.auth) |