Pythonic Integration

import dagster as dg
from dagster_dbt import DbtCliResource, DbtProject, dbt_assets

# Define the dbt project
my_dbt_project = DbtProject(project_dir="path/to/dbt_project", target="dev")

# Create the dbt resource
dbt_resource = DbtCliResource(project_dir=my_dbt_project.project_dir)


# Define assets from dbt models
@dbt_assets(manifest=my_dbt_project.manifest_path)
def my_dbt_assets(context: dg.AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()


# Define definitions
defs = dg.Definitions(assets=[my_dbt_assets], resources={"dbt": dbt_resource})

my_dbt_project = DbtProject(project_dir="path/to/dbt_project")
my_dbt_project.prepare_if_dev()  # Runs dbt deps + dbt parse if needed

@dbt_assets(
    manifest=my_dbt_project.manifest_path,
    select="tag:daily",
    exclude="tag:deprecated",
)
def my_dbt_assets(context, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

from collections.abc import Mapping
from typing import Any, Optional

import dagster as dg
from dagster_dbt import DagsterDbtTranslator, DbtProject, dbt_assets


class CustomDbtTranslator(DagsterDbtTranslator):
    def get_asset_spec(
        self,
        manifest: Mapping[str, Any],
        unique_id: str,
        project: Optional[DbtProject],
    ) -> dg.AssetSpec:
        base_spec = super().get_asset_spec(manifest, unique_id, project)
        dbt_props = self.get_resource_props(manifest, unique_id)

        return base_spec.merge_attributes(
            metadata={"model_name": dbt_props["name"]}, group_name="analytics"
        )


@dbt_assets(
    manifest=my_dbt_project.manifest_path,
    dagster_dbt_translator=CustomDbtTranslator(),
)
def my_dbt_assets(context, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

models:
  - name: customers
    meta:
      custom_field: custom_value

import json

import dagster as dg
from dagster_dbt import dbt_assets

daily_partitions = dg.DailyPartitionsDefinition(start_date="2023-01-01")


@dbt_assets(manifest=my_dbt_project.manifest_path, partitions_def=daily_partitions)
def my_dbt_assets(context: dg.AssetExecutionContext, dbt: DbtCliResource):
    time_window = context.partition_time_window

    dbt_vars = {
        "min_date": time_window.start.strftime("%Y-%m-%d"),
        "max_date": time_window.end.strftime("%Y-%m-%d"),
    }

    yield from dbt.cli(["build", "--vars", json.dumps(dbt_vars)], context=context).stream()

{{ config(materialized='incremental', unique_key='order_date') }}

select * from {{ ref('my_model') }}

{% if is_incremental() %}
where order_date >= '{{ var('min_date') }}' and order_date <= '{{ var('max_date') }}'
{% endif %}

@dbt_assets(manifest=my_dbt_project.manifest_path)
def my_dbt_assets(context, dbt: DbtCliResource):
    yield from (
        dbt.cli(["build"], context=context).stream().fetch_row_counts().fetch_column_metadata()
    )

import dagster as dg
from dagster_dbt import get_asset_key_for_model


@dg.asset(deps=[get_asset_key_for_model([my_dbt_assets], "customers")])
def export_customers():
    # Use the customers model
    pass

from dagster_dbt import build_schedule_from_dbt_selection

daily_dbt_schedule = build_schedule_from_dbt_selection(
    [my_dbt_assets],
    job_name="daily_dbt_models",
    cron_schedule="0 0 * * *",
    dbt_select="tag:daily",
)

import dagster as dg
from dagster_dbt import build_dbt_asset_selection

# Select dbt models and all downstream assets
dbt_models = build_dbt_asset_selection([my_dbt_assets], dbt_select="tag:daily")
all_assets = dbt_models.downstream(include_self=True)

daily_job = dg.define_asset_job(name="daily_pipeline", selection=all_assets)
daily_schedule = dg.ScheduleDefinition(job=daily_job, cron_schedule="0 0 * * *")

class AutomatedDbtTranslator(DagsterDbtTranslator):
    def get_asset_spec(self, manifest, unique_id, project) -> dg.AssetSpec:
        base_spec = super().get_asset_spec(manifest, unique_id, project)
        return base_spec.replace_attributes(automation_condition=dg.AutomationCondition.eager())

import dagster as dg
from dagster_dbt import dbt_assets


class DbtConfig(dg.Config):
    full_refresh: bool = False


@dbt_assets(manifest=my_dbt_project.manifest_path)
def my_dbt_assets(context, dbt: DbtCliResource, config: DbtConfig):
    args = ["build"]
    if config.full_refresh:
        args.append("--full-refresh")

    yield from dbt.cli(args, context=context).stream()