def start_orders_import_job()

in WorkflowJobs/importDatasetJob.py [0:0]

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

def start_orders_import_job(s3DataPath, datasetName, datasetGroupArn, role_arn):
    # Specify the schema of your dataset here. Make sure the order of columns matches the raw data files.
    schema = {
    "Attributes": [
        {
            "AttributeName": "timestamp",
            "AttributeType": "timestamp"
        },
        {
            "AttributeName": "item_id",
            "AttributeType": "string"
        },
        {
            "AttributeName": "demand",
            "AttributeType": "integer"
        },
        {
            "AttributeName": "location",
            "AttributeType": "string"
        }
    ]
    }

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

    TargetdatasetArn = response['DatasetArn']
    workflow_params['targetTimeSeriesDataset'] = TargetdatasetArn
    updateDatasetResponse = forecast.update_dataset_group(DatasetGroupArn=datasetGroupArn, DatasetArns=[TargetdatasetArn])

    # Orders dataset import job
    datasetImportJobName = 'INVENTORY_DSIMPORT_JOB_TARGET'
    ds_import_job_response=forecast.create_dataset_import_job(DatasetImportJobName=datasetImportJobName,
                                                            DatasetArn=TargetdatasetArn,
                                                            DataSource= {
                                                                "S3Config" : {
                                                                    "Path": s3DataPathOrders,
                                                                    "RoleArn": role_arn
                                                                } 
                                                            },
                                                            TimestampFormat=TIMESTAMP_FORMAT
                                                            )

    ds_import_job_arn=ds_import_job_response['DatasetImportJobArn']

    workflow_params['ordersImportJobRunId'] = ds_import_job_arn
    
    return {
    "importJobArn": ds_import_job_arn,
    "datasetGroupArn": datasetGroupArn,
    "ordersDatasetArn": TargetdatasetArn
    }