def find_upstream_task_group()

in o2a/converter/workflow.py [0:0]

• 9 lines of code
• 4 McCabe index (conditional complexity)

    def find_upstream_task_group(self, target_task_group) -> List[TaskGroup]:
        result = []
        for task_group in self.task_groups.values():
            if (
                target_task_group.name in task_group.downstream_names
                or target_task_group.name == task_group.error_downstream_name
            ):
                result.append(task_group)
        return result