dbt cost insights with a component

component.py

from collections.abc import Iterator

import dagster as dg
from dagster_dbt import DbtCliResource, DbtProjectComponent


class DbtProjectWithInsightsComponent(DbtProjectComponent):
    """A DbtProjectComponent subclass that enables Dagster+ Insights.

    Chains `.with_insights()` onto the dbt event iterator so that warehouse
    query costs are associated with the produced asset materializations in
    the Dagster+ Insights dashboard.

    Requires:
      - dagster-cloud installed
      - Snowflake or BigQuery adapter in the dbt project
      - An InsightsSnowflakeResource (or BigQuery equivalent) configured
    """

    def execute(
        self, context: dg.AssetExecutionContext, dbt: DbtCliResource
    ) -> Iterator:
        yield from self._get_dbt_event_iterator(context, dbt).with_insights()

gistfile1.txt

type: dagster_open_platform.lib.dbt.insights_component.DbtProjectWithInsightsComponent

attributes:
  project_dir: ../../../dagster_open_platform_dbt
  select: "tag:daily"
  exclude: "config.materialized:incremental"
  include_metadata:
    - column_metadata