dagster_dbt_assets_quickstart_with_cli_resource.py

1import os
2from pathlib import Path
3
4from dagster import AssetExecutionContext, Definitions
5from dagster_dbt import DbtCliResource, dbt_assets
6
7# Point to the directory containing the dbt project
8DBT_PROJECT_DIR = Path(__file__).joinpath("..", "..", "my_dbt_project").resolve()
9
10# Create a resource to interact with the dbt CLI
11dbt = DbtCliResource(project_dir=os.fspath(DBT_PROJECT_DIR))
12
13# If the dbt project has been compiled, Dagster can read the manifest.json
14# to automatically generate asset definitions for each dbt model.
15dbt_manifest_path = (
16    DBT_PROJECT_DIR.joinpath("target", "manifest.json")
17)
18
19@dbt_assets(manifest=dbt_manifest_path)
20def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
21    # This function will run 'dbt build' for the selected assets
22    yield from dbt.cli(["build"], context=context).get_artifacts()
23
24# Define the Dagster project with the dbt assets and resource
25defs = Definitions(
26    assets=[my_dbt_assets],
27    resources={
28        "dbt": dbt,
29    },
30)