def _create_notification_task_group()

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

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

    def _create_notification_task_group(url_template, start_task_name, status, error_code=0):
        url = (
            url_template.replace("$parentId", "").replace("$jobId", "{{ run_id }}").replace("$status", status)
        )
        start_workflow_node = TaskGroup(
            name=start_task_name,
            tasks=[
                Task(
                    task_id=start_task_name,
                    template_name="http.tpl",
                    template_params=dict(url=url, error_code=error_code),
                )
            ],
        )
        return start_workflow_node