开发dbt宏遇报错：无法获取Schema内表及列信息，求解决方案

报错原因及解决 #

1. module 'dbt.exceptions' has no attribute 'RuntimeException' #

dbt没有RuntimeException这个异常类，原代码中因调用错误的Adapter方法（比如list_relations）间接触发了该错误，修正方法后异常会自动消失。若需主动抛出异常，应使用dbt.exceptions.RuntimeError或dbt.exceptions.CompilationError。

2. 'dbt.context.providers.RuntimeDatabaseWrapper object' has no attribute 'list_relations' #

dbt Adapter API中获取指定Schema下的表，正确方法是adapter.list_relations_by_prefix(schema_name)，而非adapter.list_relations。这个方法会返回Schema下所有表/视图的Relation对象。

获取Schema内表及列名的正确方式 #

• 获取表列表：用adapter.list_relations_by_prefix(schema_name)拿到指定Schema下的所有表对象（每个对象是Relation类型，包含表名、Schema等信息）。
• 获取表的列：对每个表对象，调用adapter.get_columns_in_relation(table)，返回Column对象列表，每个对象包含name（列名）、data_type（数据类型）等属性。

修正后的宏代码 #

{% macro transform_source(schema_name, source_name, table_name, alias_prefix='') %}
    {% set target_tables = adapter.list_relations_by_prefix(schema_name) %}
    {% set target_columns = [] %}

    {# 收集目标Schema下所有表的列名（小写匹配用）和原列名 #}
    {% for table in target_tables %}
        {% set columns = adapter.get_columns_in_relation(table) %}
        {% for col in columns %}
            {% set lower_col_name = col.name.lower() %}
            {% do target_columns.append((lower_col_name, col.name)) %}
        {% endfor %}
    {% endfor %}

    {# 去重目标列，避免同一列名重复处理 #}
    {% set unique_target_columns = target_columns | unique | list %}

    {# 获取源表的列信息 #}
    {% set source_rel = source(source_name, table_name) %}
    {% set source_columns = adapter.get_columns_in_relation(source_rel) %}

    {% set model_sql %}
/* ============================================= 
Home Database: 
View/Table: {{ source_name }}.{{ table_name }}
Author:
Create date: {{ run_started_at.strftime("%Y-%m-%d") }}
Description: Auto-generated column alias mapping

Revision History: 
{{ run_started_at.strftime("%Y-%m-%d") }} - Initial creation.
============================================= */ 

WITH Source AS (
    SELECT *
    FROM {{ source(source_name, table_name) }}
),

Renamed AS (
    SELECT
        {% for col in source_columns %}
            {% set lower_col = col.name.lower() %}
            {% set matched_alias = unique_target_columns | selectattr('0', 'equalto', lower_col) | map(attribute='1') | first %}
            {% if matched_alias %}
                {{ col.name }} AS {{ alias_prefix ~ matched_alias }}{% if not loop.last %},{% endif %}
            {% else %}
                {{ col.name }} AS /* TODO: Set alias for {{ col.name }} */ {% if not loop.last %},{% endif %}
            {% endif %}
        {% endfor %}
    FROM Source
)

SELECT *
FROM Renamed
    {% endset %}

    {% if execute %}
        {{ log(model_sql, info=True) }}
        {% do return(model_sql) %}
    {% endif %}
{% endmacro %}

关键修改说明 #

• 替换错误的表列表获取方法，改用adapter.list_relations_by_prefix。
• 修正列信息获取的参数，传入Relation对象而非字符串。
• 修复原代码中未初始化的变量、大小写拼写错误（比如columns_Source改为source_columns）。
• 添加列名去重逻辑，避免同一列名重复匹配。
• 优化别名匹配逻辑：用Jinja过滤器快速查找源列在目标列中的匹配项，存在则自动使用目标列名（支持加前缀），不存在则预留TODO注释位置。
• 简化源表引用，去掉不必要的raw标签，直接使用dbt的source函数。

使用方法 #

在dbt CLI中执行以下命令调用宏：

dbt run-operation transform_source --args '{schema_name: "你的目标Schema名", source_name: "dbt源定义名", table_name: "源表名", alias_prefix: "可选的别名前缀"}'

运行后会在CLI输出自动生成的SQL脚本，包含已匹配的列别名和待手动设置的占位符。

内容的提问来源于stack exchange，提问作者thesaint99