def _create_status_notification_task_group()

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

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

    def _create_status_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 = StatusNotificationTaskGroup(
            name=task_name,
            tasks=[
                Task(
                    task_id=f"{task_name}_STATUS",
                    template_name="http.tpl",
                    template_params=dict(url=url, error_code=0),
                )
            ],
        )
        return notification_tgrp