def _update_dag_tasks(function_name, caller_line, dependencies, depends_logic, args=None, template_name=None, step_name=None):
    '\n    A task in DAG of Argo YAML contains name, related template and parameters.\n    Here we insert a single task into the global tasks.\n    '
    if (step_name is None):
        function_id = utils.invocation_name(function_name, caller_line)
    else:
        function_id = step_name
    task_template = states.workflow.get_dag_task(function_id)
    if (task_template is None):
        task_template = OrderedDict({'name': function_id})
        if ((dependencies is not None) and isinstance(dependencies, list)):
            if ('dependencies' in task_template):
                task_template['dependencies'].extend(dependencies)
            else:
                task_template['dependencies'] = dependencies
        if (depends_logic is not None):
            task_template['depends'] = depends_logic
        if (template_name is None):
            task_template['template'] = function_name
        else:
            task_template['template'] = template_name
        if (args is not None):
            (parameters, artifacts) = _get_params_and_artifacts_from_args(args, function_name, prefix='tasks')
            if (len(parameters) > 0):
                task_template['arguments'] = OrderedDict()
                task_template['arguments']['parameters'] = parameters
            if (len(artifacts) > 0):
                if ('arguments' not in task_template):
                    task_template['arguments'] = OrderedDict()
                task_template['arguments']['artifacts'] = artifacts
    else:
        if (dependencies is not None):
            if ('dependencies' in task_template):
                task_template['dependencies'].extend(dependencies)
            else:
                task_template['dependencies'] = [dependencies]
        if (depends_logic is not None):
            task_template['depends'] = depends_logic
    t_name = (function_name if (template_name is None) else template_name)
    step = Step(name=function_id, template=t_name)
    if states._exit_handler_enable:
        if (states._when_prefix is not None):
            step.when = states._when_prefix
        if (function_id in states.workflow.exit_handler_step):
            states.workflow.exit_handler_step.get(function_id).append(step.to_dict())
        else:
            states.workflow.exit_handler_step[function_id] = [step.to_dict()]
    elif (states._when_prefix is not None):
        step.when = states._when_prefix
        if (step.name not in states.workflow.dag_tasks.keys()):
            step_spec = step.to_dict()
            step_spec['dependencies'] = [states._when_task]
            states.workflow.dag_tasks[step.name] = step_spec
    else:
        states.workflow.update_dag_task(function_id, task_template)
    return function_id
A task in DAG of Argo YAML contains name, related template and parameters.
Here we insert a single task into the global tasks.
def _update_dag_tasks(function_name, caller_line, dependencies, depends_logic, args=None, template_name=None, step_name=None):
    '\n    A task in DAG of Argo YAML contains name, related template and parameters.\n    Here we insert a single task into the global tasks.\n    '
    if (step_name is None):
        function_id = utils.invocation_name(function_name, caller_line)
    else:
        function_id = step_name
    task_template = states.workflow.get_dag_task(function_id)
    if (task_template is None):
        task_template = OrderedDict({'name': function_id})
        if ((dependencies is not None) and isinstance(dependencies, list)):
            if ('dependencies' in task_template):
                task_template['dependencies'].extend(dependencies)
            else:
                task_template['dependencies'] = dependencies
        if (depends_logic is not None):
            task_template['depends'] = depends_logic
        if (template_name is None):
            task_template['template'] = function_name
        else:
            task_template['template'] = template_name
        if (args is not None):
            (parameters, artifacts) = _get_params_and_artifacts_from_args(args, function_name, prefix='tasks')
            if (len(parameters) > 0):
                task_template['arguments'] = OrderedDict()
                task_template['arguments']['parameters'] = parameters
            if (len(artifacts) > 0):
                if ('arguments' not in task_template):
                    task_template['arguments'] = OrderedDict()
                task_template['arguments']['artifacts'] = artifacts
    else:
        if (dependencies is not None):
            if ('dependencies' in task_template):
                task_template['dependencies'].extend(dependencies)
            else:
                task_template['dependencies'] = [dependencies]
        if (depends_logic is not None):
            task_template['depends'] = depends_logic
    t_name = (function_name if (template_name is None) else template_name)
    step = Step(name=function_id, template=t_name)
    if states._exit_handler_enable:
        if (states._when_prefix is not None):
            step.when = states._when_prefix
        if (function_id in states.workflow.exit_handler_step):
            states.workflow.exit_handler_step.get(function_id).append(step.to_dict())
        else:
            states.workflow.exit_handler_step[function_id] = [step.to_dict()]
    elif (states._when_prefix is not None):
        step.when = states._when_prefix
        if (step.name not in states.workflow.dag_tasks.keys()):
            step_spec = step.to_dict()
            step_spec['dependencies'] = [states._when_task]
            states.workflow.dag_tasks[step.name] = step_spec
    else:
        states.workflow.update_dag_task(function_id, task_template)
    return function_id