def create_job()

in src/aws_interfaces/iot_interface.py [0:0]

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

    def create_job(self, deployConfig):
        logging.info('create_job jobId %s', deployConfig['defaultConfig']['jobId'])
        kwargs = {
            'jobId': deployConfig['defaultConfig']['jobId'],
            'targets': deployConfig['defaultConfig']['thingArnList'],
            'documentSource': deployConfig['defaultConfig']['jobDocumentSrc'],
            'targetSelection': deployConfig['defaultConfig']['targetSelection']
        }

        if 'presignedUrlConfig' in deployConfig:
            kwargs['presignedUrlConfig'] = deployConfig['presignedUrlConfig']
        if 'jobExecutionsRolloutConfig' in deployConfig:
            kwargs['jobExecutionsRolloutConfig'] = deployConfig['jobExecutionsRolloutConfig']
        if 'abortConfig' in deployConfig:
            kwargs['abortConfig'] = deployConfig['abortConfig']
        if 'timeoutConfig' in deployConfig:
            kwargs['timeoutConfig'] = deployConfig['timeoutConfig']

        try:
            response = self.client.create_job(**kwargs)
        except ClientError as e:
            return False, str(e)
        return True, None