get_relationship_values.sql

{%- macro get_relationship_values(node_type) -%}
    {{ return(adapter.dispatch('get_relationship_values', 'dbt_project_evaluator')(node_type)) }}
{%- endmacro -%}

{%- macro default__get_relationship_values(node_type) -%}

    {%- if execute -%}
        {%- if node_type == 'nodes' %}
            {% set nodes_list = graph.nodes.values() %}   
        {%- elif node_type == 'exposures' -%}
            {% set nodes_list = graph.exposures.values() %}
        {%- elif node_type == 'metrics' -%}
            {% set nodes_list = graph.metrics.values() %}
        {%- else -%}
            {{ exceptions.raise_compiler_error("node_type needs to be either nodes, exposures or metrics, got " ~ node_type) }}
        {% endif -%}
        
        {%- set values = [] -%}

        {%- for node in nodes_list -%}

            {%- if node.get('depends_on',{}).get('nodes',[]) |length == 0 -%}

                {%- set values_line =
                  [
                    "cast('" ~ node.unique_id ~ "' as " ~ dbt_project_evaluator.type_string_dpe() ~ ")",
                    "cast(NULL as " ~ dbt_project_evaluator.type_string_dpe() ~ ")",
                    "cast(" ~ dbt_project_evaluator.bool_literal(false) | trim ~ " as " ~ dbt.type_boolean() ~ ")",
                  ]
                %}
                  
                {%- do values.append(values_line) -%}

            {%- else -%}       

                {%- for parent in node.get('depends_on',{}).get('nodes',[]) -%}

                    {# Use attached_node to determine primary relationship for tests (works in both Core and Fusion) #}
                    {# Fall back to loop.last for backwards compatibility if attached_node is not available #}
                    {%- set is_primary = (parent == node.get('attached_node')) if node.get('attached_node') else loop.last -%}
                    {%- set values_line =
                        [
                            "cast('" ~ node.unique_id ~ "' as " ~ dbt_project_evaluator.type_string_dpe() ~ ")",
                            "cast('" ~ parent ~ "' as " ~ dbt_project_evaluator.type_string_dpe() ~ ")",
                            "cast(" ~ dbt_project_evaluator.bool_literal(is_primary if node.unique_id.split('.')[0] == 'test' else false) | trim ~ " as " ~ dbt.type_boolean() ~ ")"
                        ]
                    %}
                      
                    {%- do values.append(values_line) -%}

                {%- endfor -%}

            {%- endif -%}

        {%- endfor -%}
    
    {{ return(values) }}

    {%- endif -%}
  
{%- endmacro -%}