1 # ============LICENSE_START=======================================================
2 # Copyright (C) 2021 Nordix Foundation
3 # ===============================================================================
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
15 # SPDX-License-Identifier: Apache-2.0
16 # ============LICENSE_END========================================================
19 from io import BytesIO
20 from pathlib import Path
24 from sdcBePy.common import logger
25 from sdcBePy.common.normative.main import process_element_list, process_type_list
26 from sdcBePy.tosca.models.normativeElementsList import get_normative_element_candidate_list, get_normative_element_with_metadata_list
30 def __init__(self, sdc_be_proxy):
31 self.__base_path = Path('/sdc2/rest/v1/catalog/model')
32 self.__imports_path = self.__base_path / 'imports'
33 self.__sdc_be_proxy = sdc_be_proxy
35 def create_model(self, model_payload_dict, model_imports_zip_path):
36 model_name = model_payload_dict['name']
38 logger.debug("Starting to create model '{}', zip path '{}'".format(model_name, model_imports_zip_path))
40 multi_part_form_data = []
42 model_zip_param = ('modelImportsZip', (pycurl.FORM_FILE, str(model_imports_zip_path)))
43 multi_part_form_data.append(model_zip_param)
45 json_payload = self.__parse_to_json_str(model_payload_dict)
46 model_param = ('model', (
47 pycurl.FORM_CONTENTS, json_payload,
48 pycurl.FORM_CONTENTTYPE, 'application/json'
50 multi_part_form_data.append(model_param)
52 response_buffer = BytesIO()
53 response_code = self.__sdc_be_proxy.post_file(str(self.__base_path), multi_part_form_data, response_buffer)
54 logger.debug("Create model response code '{}'".format(response_code))
55 if response_code != 201:
56 error_msg = "Failed to create model '{}'".format(model_name)
57 logger.log(error_msg, response_buffer.getvalue())
58 raise Exception(error_msg)
59 logger.log("Created model", model_name)
61 def update_model_imports(self, model_payload_dict, model_imports_zip_path):
62 model_name = model_payload_dict['name']
63 logger.debug("Starting to update model '{}', zip path '{}'".format(model_name, model_imports_zip_path))
65 multi_part_form_data = []
67 model_zip_post = ('modelImportsZip', (pycurl.FORM_FILE, str(model_imports_zip_path)))
68 multi_part_form_data.append(('modelName', model_name))
69 multi_part_form_data.append(model_zip_post)
71 response_buffer = BytesIO()
72 response_code = self.__sdc_be_proxy.put_file(str(self.__imports_path), multi_part_form_data, response_buffer)
73 logger.debug("Update model response code '{}'".format(response_code))
74 if response_code != 204:
75 error_msg = "Failed to update model '{}'".format(model_name)
76 logger.log(error_msg, response_buffer.getvalue())
77 raise Exception(error_msg)
78 logger.log("Updated model", model_name)
80 def import_model_elements(self, model_payload_dict, tosca_elements_import_path, with_metadata=False):
81 model_name = model_payload_dict['name']
82 logger.debug("Starting import of normative elements for model '{}'".format(model_name))
84 process_element_list(get_normative_element_with_metadata_list(tosca_elements_import_path), self.__sdc_be_proxy, model=model_name)
86 process_element_list(get_normative_element_candidate_list(tosca_elements_import_path), self.__sdc_be_proxy, model=model_name)
87 logger.log("Finished importing normative elements for model", model_name)
89 def import_model_types(self, model_payload_dict, types_list, upgrade):
90 model_name = model_payload_dict['name']
91 logger.debug("Starting import of normative types for model '{}'".format(model_name))
92 process_type_list(types_list, self.__sdc_be_proxy, upgrade)
93 logger.log("Finished importing normative types for model", model_name)
96 def __parse_to_json_str(model_payload_dict):
97 return json.dumps(model_payload_dict)