def apply_pipeline_defaults()

in liminal/core/config/defaults/default_configs.py [0:0]

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

def apply_pipeline_defaults(subliminal, superliminal, pipeline):
    """Apply defaults values on given pipeline
    :param subliminal: subliminal config
    :param superliminal: superliminal config
    :param  pipeline: to apply defaults on

    :returns: enriched pipeline with superliminal.pipeline_defaults & subliminal.pipeline_defaults
    """
    keyword = "pipeline_defaults"
    superliminal_pipe_defaults = superliminal.get(keyword, {}).copy()
    subliminal_pipe_defaults = subliminal.get(keyword, {}).copy()
    superliminal_before_tasks = superliminal_pipe_defaults.pop(__BEFORE_TASKS, [])
    superliminal_after_tasks = superliminal_pipe_defaults.pop(__AFTER_TASKS, [])
    merged_pipeline_defaults = merge_dicts(subliminal_pipe_defaults, superliminal_pipe_defaults, True)
    pipeline = merge_dicts(pipeline, merged_pipeline_defaults, True)

    return apply_task_defaults(
        subliminal,
        superliminal,
        pipeline,
        superliminal_before_tasks=superliminal_before_tasks,
        superliminal_after_tasks=superliminal_after_tasks,
    )