catalog-be/src/main/resources/scripts/sdcBePy/tosca/imports/run.py

   1 #!/usr/bin/env python3
   2
   3 import os
   4 from pathlib import Path
   5
   6 import sdcBePy.common.logger as logger
   7 from sdcBePy.common.normative.main import process_element_list, process_type_list
   8 from sdcBePy.tosca.main import parse_and_create_proxy
   9 from sdcBePy.tosca.models import normativeElementsList
  10 from sdcBePy.tosca.models import normativeTypesList
  11 from sdcBePy.tosca.models.model_client import ModelClient
  12 from sdcBePy.tosca.models.model_import_manager import ModelImportManager
  13 from sdcBePy.tosca.models.node_type_client import NodeTypeClient
  14
  15
  16 def main(sdc_be_proxy, update_version):
  17     # use to run script form this dir (not like the command)
  18     # base_file_location = os.getcwd() + "/../../../import/tosca/"
  19     base_file_location = os.getcwd() + os.path.sep
  20     logger.debug("working directory =" + base_file_location)
  21     if sdc_be_proxy.disable_locking("true") != 200:
  22         raise RuntimeError("Failed to disable locking")
  23     try:
  24         process_element_list(normativeElementsList.get_normative_element_candidate_list(base_file_location), sdc_be_proxy)
  25         process_type_list(normativeTypesList.get_normative_type_candidate_list(base_file_location), sdc_be_proxy, update_version)
  26         process_element_list(normativeElementsList.get_normative_element_with_metadata_list(base_file_location), sdc_be_proxy)
  27         # Add model based normatives
  28         model_import_manager = ModelImportManager(Path(base_file_location) / 'models', ModelClient(sdc_be_proxy),
  29                                                   NodeTypeClient(sdc_be_proxy))
  30         model_import_manager.deploy_models()
  31     except Exception as ex:
  32         logger.log("An error has occurred while uploading elements and types: ", str(ex))
  33         raise ex
  34     finally:
  35         if sdc_be_proxy.disable_locking("false") != 200:
  36             raise RuntimeError("Failed to enable locking")
  37     logger.log("Script end ->", "All normatives imported successfully!")
  38     logger.print_and_exit(0, None)
  39
  40
  41 def run():
  42     sdc_be_proxy, update_version = parse_and_create_proxy()
  43     main(sdc_be_proxy, update_version)
  44
  45
  46 if __name__ == "__main__":
  47     run()