def __calculate_job_result()

in liminal/runners/airflow/operators/job_status_operator.py [0:0]

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

    def __calculate_job_result(self, context):
        self.log.info('scanning task instances states.. ')
        task_instances = context['dag_run'].get_task_instances()
        task_states = [
            self.__log_and_get_state(task_instance)
            for task_instance in task_instances
            if task_instance.task_id != context['task_instance'].task_id
        ]

        self.__job_result = 0
        if all((state == State.SUCCESS or state == State.SKIPPED) for state in task_states):
            self.__job_result = 1