[dagster-dbt] add with_snowflake/bigquery_insights chainable method

[benpankow]

Summary

Adds a with_snowflake_insights and with_bigquery_insights methods that can be chained to a .stream() call, simplifying using insights alongside other deferred fetches:

Before:

@dbt_assets(manifest=snowflake_manifest_path)
def jaffle_shop_dbt_assets(
    context: AssetExecutionContext,
    dbt: DbtCliResource,
):
    dbt_cli_invocation = dbt.cli(["build", "--profile", "unittest"], context=context)
    yield from dbt_with_snowflake_insights(context, dbt_cli_invocation)

Now:

@dbt_assets(manifest=snowflake_manifest_path)
def jaffle_shop_dbt_assets(
    context: AssetExecutionContext,
    dbt: DbtCliResource,
):
    yield from dbt.cli(
        ["build", "--profile", "unittest"], context=context
    ).stream().with_snowflake_insights()

Test Plan

https://github.com/dagster-io/internal/pull/10731

[benpankow]

• [dagster-dbt] add with_snowflake/bigquery_insights chainable method #23183 👈
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @benpankow and the rest of your teammates on Graphite

[rexledesma]

Can we just have one function, with_insights? We can determine whether the target is snowflake or bigquery from the execution manifest

Also, instead of an import time error, maybe we could raise a warning instead, so core execution doesn't fail? We should also add a test in the OSS context (the warning is thrown since dagster-cloud isn't installed)

[benpankow]

Can we just have one function, with_insights? We can determine whether the target is snowflake or bigquery from the execution manifest

Good call; updated.

Also, instead of an import time error, maybe we could raise a warning instead, so core execution doesn't fail? We should also add a test in the OSS context (the warning is thrown since dagster-cloud isn't installed)

My bias is for the code to explicitly error here just so there's no silent failure or other unexpected behavior - in the case that a user is conditionally using different warehouses, I think they can just conditionally call this method. They'd have to do a fair amount of conditional logic (manifest, target) elsewhere anyway.

qu

[rexledesma]

Seems like we should just universally have the config check enabled? That way, we can remove it as an argument. But we should probably ensure that check is done at the start of the dbt cli invocation so that things fail fast.

Not a blocker to get this initial scaffold in though