def _create_transition_notification_task_group()

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

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

    def _create_transition_notification_task_group(url_template, task_name, node_name, status):
        url = (
            url_template.replace("$jobId", "{{ dag.dag_id }}")
            .replace("$nodeName", node_name)
            .replace("$status", status)
        )
        notification_tgrp = TransitionNotificationTaskGroup(
            name=task_name,
            tasks=[
                Task(
                    task_id=f"{task_name}_TRANSITION",
                    template_name="http.tpl",
                    template_params=dict(url=url, error_code=0),
                )
            ],
        )
        return notification_tgrp