def _add_transition_and_status()

in o2a/transformers/add_node_notificaton_transformer.py [0:0]

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

    def _add_transition_and_status(self, downstream_name, task_group, workflow):
        transition_notification = self._create_transition_notification_task_group(
            self.notification_url,
            f"{task_group.name}{NODE_TRANSITION_SUFFIX}_T_{downstream_name}",
            task_group.name,
            f"T:{downstream_name}",
        )
        status_notification = self._create_status_notification_task_group(
            self.notification_url, f"{downstream_name}{NODE_STATUS_SUFFIX}", downstream_name, "S:RUNNING"
        )
        workflow.task_groups[transition_notification.name] = transition_notification
        workflow.task_groups[status_notification.name] = status_notification
        task_group.downstream_names.remove(downstream_name)
        task_group.downstream_names.append(transition_notification.name)
        transition_notification.downstream_names = [status_notification.name]
        status_notification.downstream_names = [downstream_name]