def update_datawrangler_source()

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

• 11 lines of code
• 4 McCabe index (conditional complexity)

    def update_datawrangler_source(self, flow, data_uri) :
        
        for node in flow["nodes"]:
            if "dataset_definition" in node["parameters"]:
                data_def = node["parameters"]["dataset_definition"]
                source_type = data_def["datasetSourceType"]

                if source_type == "S3":
                    data_def["name"] = "input_data"
                    node["parameters"]["dataset_definition"]["s3ExecutionContext"]["s3Uri"] = data_uri
                    self.persist_flow()
                else:
                    raise ValueError(f"{source_type} support has not been implemented for this blueprint.")