simplifying file, reducing code

Author: digitalghost-dev

card_data/pipelines/definitions.py

@@ -1,7 +1,6 @@
 from pathlib import Path
 
 from dagster import definitions, load_from_defs_folder
-from dagster_dbt import DbtCliResource, DbtProject, dbt_assets
 
 import dagster as dg
 
@@ -11,31 +10,19 @@
 
 @definitions
 def defs():
+    # load_from_defs_folder discovers dbt assets from transform_data.py
     folder_defs = load_from_defs_folder(project_root=Path(__file__).parent.parent)
     return dg.Definitions.merge(folder_defs, defs_pricing)
 
-dbt_project_directory = Path(__file__).absolute().parent / "poke_cli_dbt"
-dbt_project = DbtProject(project_dir=dbt_project_directory)
-
-dbt_resource = DbtCliResource(project_dir=dbt_project)
-
-# Compiles the dbt project & allow Dagster to build an asset graph
-dbt_project.prepare_if_dev()
-
-# Yields Dagster events streamed from the dbt CLI
-@dbt_assets(manifest=dbt_project.manifest_path)
-def dbt_models(context: dg.AssetExecutionContext, dbt: DbtCliResource):
-    yield from dbt.cli(["build"], context=context).stream()
-
-# Define the pricing pipeline job that materializes both assets
+# Define the pricing pipeline job that materializes the assets and downstream dbt model
 pricing_pipeline_job = dg.define_asset_job(
     name="pricing_pipeline_job",
-    selection=dg.AssetSelection.assets(build_dataframe, load_pricing_data),
+    selection=dg.AssetSelection.assets(build_dataframe, load_pricing_data).downstream(include_self=True),
 )
 
 price_schedule = dg.ScheduleDefinition(
     name="price_schedule",
-    cron_schedule="10 10 * * *",
+    cron_schedule="31 21 * * *",
     target=pricing_pipeline_job,
     execution_timezone="America/Los_Angeles",
 )