def start_products_import_job()

in WorkflowJobs/importDatasetJob.py [0:0]

• 40 lines of code
• 1 McCabe index (conditional complexity)

def start_products_import_job(s3DataPath, datasetName, datasetGroupArn, role_arn, ordersDatasetArn):
        # Specify the schema of your dataset here. Make sure the order of columns matches the raw data files.
    schema = {
        "Attributes": [
            {
                "AttributeName": "item_id",
                "AttributeType": "string"
            },
            {
                "AttributeName": "description",
                "AttributeType": "string"
            },
            {
                "AttributeName": "price",
                "AttributeType": "float"
            }
        ]
    }

    response = forecast.create_dataset(
                    Domain="INVENTORY_PLANNING",
                    DatasetType='ITEM_METADATA',
                    DatasetName=datasetName+'_2',
                    DataFrequency=DATASET_FREQUENCY, 
                    Schema = schema)

    metaDatasetArn = response['DatasetArn']
    workflow_params['itemMetaDataset'] = metaDatasetArn
    updateDatasetResponse = forecast.update_dataset_group(DatasetGroupArn=datasetGroupArn, DatasetArns=[ordersDatasetArn, metaDatasetArn])

    # Dataset import job
    datasetImportJobName = 'INVENTORY_DSIMPORT_JOB_METADATA'
    ds_import_job_response=forecast.create_dataset_import_job(DatasetImportJobName=datasetImportJobName,
                                                          DatasetArn=metaDatasetArn,
                                                          DataSource= {
                                                              "S3Config" : {
                                                                 "Path": s3DataPathProducts,
                                                                 "RoleArn": role_arn
                                                              } 
                                                          },
                                                          TimestampFormat=TIMESTAMP_FORMAT
                                                         )

    ds_import_job_arn=ds_import_job_response['DatasetImportJobArn']

    workflow_params['productsImportJobRunId'] = ds_import_job_arn
    return