def modeltransform_start()

in src/graph_notebook/magics/ml.py [0:0]

• 63 lines of code
• 19 McCabe index (conditional complexity)

def modeltransform_start(args: argparse.Namespace, client: Client, params):
    """
    Starts a new modeltransform job. If Params is not empty, we will attempt to parse it into JSON
    and use it as the command payload. Otherwise we will check args for the required parameters:
    """

    if params is None or params == '' or params == {}:
        data = {
            'id': args.job_id
        }
        if args.base_processing_instance_type:
            data['baseProcessingInstanceType'] = args.base_processing_instance_type
        if args.base_processing_instance_volume_size_in_gb:
            data['baseProcessingInstanceVolumeSizeInGB'] = args.base_processing_instance_volume_size_in_gb
        data = add_security_params(args, data)
        s3_output_uri = args.s3_output_uri
        data_processing_job_id = args.data_processing_job_id
        model_training_job_id = args.model_training_job_id
        training_job_name = args.training_job_name
    else:
        if type(params) is dict:
            data = params
        else:
            try:
                data = json.loads(params)
            except ValueError:
                print("Error: Unable to load modeltransform parameters. Please check that they are defined in JSON "
                      "format.")
        if 'modeltransform' in data:
            data = data['modeltransform']
        if 'modelTransformOutputS3Location' in data:
            s3_output_uri = data['modelTransformOutputS3Location']
        else:
            s3_output_uri = args.s3_output_uri
        has_dataprocessing_id = False
        has_training_id = False
        has_training_name = False
        try:
            if 'dataProcessingJobId' in data:
                data_processing_job_id = data['dataProcessingJobId']
            else:
                data_processing_job_id = args.data_processing_job_id
            has_dataprocessing_id = True
        except AttributeError:
            pass
        try:
            if 'mlModelTrainingJobId' in data:
                model_training_job_id = data['mlModelTrainingJobId']
            else:
                model_training_job_id = args.model_training_job_id
            has_training_id = True
        except AttributeError:
            pass
        try:
            if 'trainingJobName' in data:
                training_job_name = data['trainingJobName']
            else:
                training_job_name = args.training_job_name
            has_training_name = True
        except AttributeError:
            pass
        if not (has_dataprocessing_id and has_training_id) and not has_training_name:
            print("You are required to define either a) dataProcessingJobId AND mlModelTrainingJobId or "
                  "b) trainingJobName as arguments when creating a transform job.")

    res: Response = client.modeltransform_create(s3_output_uri, data_processing_job_id,
                                                 model_training_job_id, training_job_name, **data)
    res.raise_for_status()
    return res.json()