如何在DBT中安全实现try/catch？GCP项目连接异常处理

方法1：通过BigQuery元数据预判断项目是否存在 #

BigQuery的INFORMATION_SCHEMA.CATALOGS视图会列出当前用户有权访问的所有GCP项目（对应BigQuery中的catalog）。先查询该视图判断目标项目是否存在，再决定是否执行后续的JOBS查询，从根源避免报错。

示例代码：

{% set target_project = "your-target-project-id" %}

{% set check_project_query %}
SELECT 1
FROM INFORMATION_SCHEMA.CATALOGS
WHERE CATALOG_NAME = '{{ target_project }}'
{% endset %}

{% set project_exists = run_query(check_project_query) %}

{% if execute %}
    {% set project_exists_flag = project_exists.rows | length > 0 %}
    {% if project_exists_flag %}
        {% set connection_query %}
        SELECT *
        FROM `{{ target_project }}`.`region-eu`.INFORMATION_SCHEMA.JOBS
        {% endset %}
        {% set result = run_query(connection_query) %}
        -- 在这里添加查询结果的处理逻辑
    {% else %}
        -- 项目不存在/无权限时的处理逻辑
        {% do log("项目 " ~ target_project ~ " 不存在或当前用户无访问权限", info=True) %}
    {% endif %}
{% endif %}

说明 #

• INFORMATION_SCHEMA.CATALOGS返回的是当前用户可访问的所有项目，无匹配记录时，统一按「项目不存在/无权限」处理即可。
• 先执行存在性检查，避免直接访问无效项目触发报错中断流程。

方法2：用Jinja错误捕获处理查询异常（DBT 1.0+支持） #

如果需要直接尝试查询，同时处理各类错误场景，可以用Jinja2的try-except块捕获run_query抛出的异常，确保流程不会中断。

示例代码：

{% set target_project = "your-target-project-id" %}

{% set connection_query %}
SELECT *
FROM `{{ target_project }}`.`region-eu`.INFORMATION_SCHEMA.JOBS
{% endset %}

{% if execute %}
    {% set result = none %}
    {% try %}
        {% set result = run_query(connection_query) %}
        -- 正常查询后的业务逻辑
    {% except %}
        -- 捕获到错误时的处理逻辑（项目不存在、权限不足等情况通用）
        {% do log("访问项目 " ~ target_project ~ " 失败：" ~ exc_info, info=True) %}
    {% endtry %}
{% endif %}

说明 #

• exc_info变量会携带异常的详细信息，可用于日志记录或问题排查。
• 该方法适合需要直接尝试查询，同时兼容多种错误场景的需求。

注意事项 #

• 确保执行DBT的账号拥有INFORMATION_SCHEMA.CATALOGS的查询权限，BigQuery默认用户均具备该权限。
• 批量处理多个项目时，可将项目ID存入列表循环遍历，每个项目单独做存在性检查或错误捕获。

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