def run()

in luigi/contrib/ecs.py [0:0]

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

    def run(self):
        if (not self.task_def and not self.task_def_arn) or \
                (self.task_def and self.task_def_arn):
            raise ValueError(('Either (but not both) a task_def (dict) or'
                              'task_def_arn (string) must be assigned'))
        if not self.task_def_arn:
            # Register the task and get assigned taskDefinition ID (arn)
            response = client.register_task_definition(**self.task_def)
            self.task_def_arn = response['taskDefinition']['taskDefinitionArn']

        run_task_kwargs = self.run_task_kwargs
        run_task_kwargs.update({
            'taskDefinition': self.task_def_arn,
            'cluster': self.cluster,
            'overrides': self.combined_overrides,
        })

        # Submit the task to AWS ECS and get assigned task ID
        # (list containing 1 string)
        response = client.run_task(**run_task_kwargs)

        if response['failures']:
            raise Exception(", ".join(["fail to run task {0} reason: {1}".format(failure['arn'], failure['reason'])
                                       for failure in response['failures']]))

        self._task_ids = [task['taskArn'] for task in response['tasks']]

        # Wait on task completion
        _track_tasks(self._task_ids, self.cluster)