def get_config()

in code/workflow/implementations/autopilot/bp_init_stage.py [0:0]

• 49 lines of code
• 6 McCabe index (conditional complexity)

    def get_config(cls, event):
        
        if cls._instance is None:
            cls._instance = cls.__new__(cls)

        config_uri = event["Input"]["config_uri"]
        config = get_json_from_s3(config_uri)

        if config :

            cls.dict = config

            if "s3_bucket" in config["workspace-config"] :
                cls.w_bucket = config["workspace-config"]["s3_bucket"]
            else :
                cls.w_bucket = event["default_workspace"]
                cls.dict["workspace-config"]["s3_bucket"] = event["default_workspace"]
            
            cls.w_prefix = config["workspace-config"]["s3_prefix"]

            now = strftime("%Y-%m-%d-%H-%M-%S", gmtime())
            if "s3_bucket" in config["data-config"] :
                cls.d_bucket = config["data-config"]["s3_bucket"]
            else :
                cls.d_bucket = event["default_workspace"]
                cls.dict["data-config"]["s3_bucket"] = event["default_workspace"]

                cls.raw_in_prefix = config["data-config"]["raw_in_prefix"]
                cls.prepped_out_prefix = config["data-config"]["prepped_out_prefix"]

                cls.flow_name = config["dataprep-config"]["definition_file"]
                cls.output_node_id = config["dataprep-config"]["output_node_id"]
                cls.dp_instance_type = config["dataprep-config"]["instance_type"]
                cls.dp_instance_count = int(config["dataprep-config"]["instance_count"])
                cls.dp_data_version = config["dataprep-config"]["data_version"]

                cls.automl_base_job_name = config["automl-config"]["job_base_name"]
                cls.automl_job_name = f"{cls.automl_base_job_name}-{now}"
                cls.automl_max_candidates = int(config["automl-config"]["max_candidates"])
                cls.automl_target_name = config["automl-config"]["target_name"]
                cls.automl_problem_type = config["automl-config"]["problem_type"]
                cls.automl_metric_name = config["automl-config"]["metric_name"]  
                
            if "iam_role" in config["security-config"] :
                cls.iam_role = config["security-config"]["iam_role"]
            else :
                cls.iam_role = event["sm_execution_role"]
                cls.dict["security-config"]["iam_role"] = event["sm_execution_role"]

            cls.model_base_name = config["model-config"]["model_base_name"]
                
            cls.bias_base_name = config["bias-analysis-config"]["job_base_name"]
            cls.xai_base_name = config["xai-config"]["job_base_name"]
            cls.error_base_name = config["error-analysis-config"]["job_base_name"]
                
            ##resource names are initialized here
            cls.dict["automl-config"]["job_name"] = cls.automl_job_name
            cls.dict["model-config"]["model_name"] = f"{cls.model_base_name}-{now}"
            cls.dict["bias-analysis-config"]["job_name"] = f"{cls.bias_base_name}-{now}"
            cls.dict["xai-config"]["job_name"] = f"{cls.xai_base_name}-{now}"
            cls.dict["error-analysis-config"]["job_name"] = f"{cls.error_base_name}-{now}"
    
        else :
            raise Exception("Failed to read config.json.")
                
        return cls._instance