dbt-f2dq/macros/meta/get_monitored.sql

{% macro pub_monitored_from_graph() %}
    {% set monitored = [] %}
    {% set both = []%}
    {% do both.extend(graph.nodes.values()) %}
    {% do both.extend(graph.sources.values()) %}
    {% set owners_config = f2dq.get_owners_config() %}

    {% set select_var = f2dq.f2dq_var('select', none) %}
    {% set select_all = true %}

    {% set selected_nodes = none %}
    {% set selected_tags = none %}

    {% if select_var is not none %}
        {% set select_all = false %}
        {% set selected_nodes = dict() %}
        {% set selected_tags = dict() %}

        {% for el in select_var %}
            {% if el.startswith('tag:') %}
                {% do selected_tags.update({el[4:]: True}) %}
            {% else %}
                {% do selected_nodes.update({el: True}) %}
            {% endif %}
        {% endfor %}
    {% endif %}

    {% for el in both %}
        {% if el.resource_type in ['model', 'seed', 'source'] %}
            {% if f2dq.f2dq_node_config(el, 'monitored') %}
                {% set target_name = el.identifier or el.alias or el.name %}

                {% if select_all %}
                    {% set selected = true %}
                {% else %}
                    {% set selected_name = selected_nodes.get(target_name, false) %}
                    {% set selected_tag = [] %}

                    {% for tag in el.tags %}
                        {% if selected_tags.get(tag, false) %}
                            {% do selected_tag.append(true) %}
                        {% endif %}
                    {% endfor %}
                    
                    {% set selected = selected_name or (selected_tag | length > 0) %}
                {% endif %}

                {% set metrics_groups = f2dq.f2dq_node_config(el, 'metrics_groups', f2dq.f2dq_var('default_metrics')) %}
                {% set additional_metrics = f2dq.f2dq_node_config(el, 'metrics', {}) %}

                {% do monitored.append({
                    'name': f2dq.name_in_db(target_name),
                    'schema': f2dq.name_in_db(el.schema),
                    'database': f2dq.name_in_db(el.database),
                    'time_filter': f2dq.f2dq_node_config(el, 'time_filter', none),
                    'metrics_groups': metrics_groups,
                    'additional_metrics': f2dq.metrics_in_db(additional_metrics),
                    'metrics': f2dq.metrics_in_db(f2dq.final_metrics(metrics_groups, additional_metrics)),
                    'columns': f2dq.columns_in_db(f2dq.f2dq_node_config(el, 'columns', none)),
                    'anomaly_detector': f2dq.f2dq_node_config(el, 'anomaly_detector', f2dq.f2dq_var('anomaly_detector', {})),
                    'owners': f2dq.prepare_model_owners(f2dq.f2dq_node_config(el, 'owners', []), owners_config),
                    'selected': selected
                    })
                %}
            {% endif %}
        {% endif %}
    {% endfor %}

    {{ return(monitored) }}
{% endmacro %}

{% macro get_owners_config() %}
    {% set owners_config = f2dq.f2dq_var('owners_config', {}) %}
    {{ return (owners_config) }}
{% endmacro %}

{% macro prepare_model_owners(f2dq_owners, owners_config) %}
    {% set owners = {} %}
    {% set seen_identifiers = {} %}
    {% for owner in f2dq_owners if owners_config.get(owner) %}
        {% set members = owners_config.get(owner) %}
        {% for member in members %}
            {% set identifier = member.get('identifier') %}
            {% if identifier not in seen_identifiers %}
            {% do seen_identifiers.update({identifier: true }) %}
            {% do owners.update({
                identifier: {
                    'notify_channel': member.get('type'),
                    'owner': owner,
                    'name': member.get('name') 
                } 
            }) %}
            {% endif %}
        {% endfor %}
    {% endfor %}
    {{ return (owners) }}
{% endmacro %}