Hi all I want to run dbt with a vars This is my code ```DBT dagster #ask-community

Hi all, I want to run dbt with a vars. This is my ...

Julius

03/22/2023, 6:37 AM

Hi all, I want to run dbt with a vars. This is my code

Copy code

DBT_PROJECT_PATH = file_relative_path(__file__, "../../../../../dbt/projects/dbt-vault")
DBT_PROFILES = file_relative_path(__file__, "../../../../../dbt/projects/dbt-vault/config")
select_schema = "*"
partitions_def = DailyPartitionsDefinition(start_date=datetime(2023, 3, 10))
dbt_assets = with_resources(
    load_assets_from_dbt_project(
        project_dir=DBT_PROJECT_PATH,
        profiles_dir=DBT_PROFILES,
        select=select_schema,
        partitions_def=partitions_def,
        key_prefix=["dbt_vault"],
    ),
    {
        "dbt": dbt_cli_resource.configured(
            {
                "project_dir": DBT_PROJECT_PATH,
                "profiles_dir": DBT_PROFILES,
                "vars": {"incremental_load_date": f"{partitions_def}"},   <---------- running with vars change by the date of partitions
            },
        )
    },
)

I want to do something like when the job start running using the partitions at: 2023-03-10 -------> "vars": {"incremental_load_date": "2023-03-10"} 2023-03-11 -------> "vars": {"incremental_load_date": "2023-03-11"} . . . so on. How can I set that up?

Julius

03/22/2023, 8:34 AM

ok, I know how to set that up!

Copy code

def vars_for_dbt(input_str: str) -> Mapping[str, Any]:
    # Process input_str and create the output dictionary for needed dbt vars
    output_dict = {
        "incremental_load_date": f"{input_str}",
    }
    return output_dict

DBT_PROJECT_PATH = file_relative_path(__file__, "../../../../../dbt/projects/dbt-vault")
DBT_PROFILES = file_relative_path(__file__, "../../../../../dbt/projects/dbt-vault/config")
exclude_model = "dbt_metrics_default_calendar"
partitions_def = DailyPartitionsDefinition(
    start_date=datetime(2023, 3, 10)
)

dbt_assets = with_resources(
    load_assets_from_dbt_project(
        project_dir=DBT_PROJECT_PATH,
        profiles_dir=DBT_PROFILES,
        exclude=exclude_model,
        partitions_def=partitions_def,      
        partition_key_to_vars_fn=vars_for_dbt,     <----- Adding this partition_key_to_vars_fn
        key_prefix=["dbt_vault"],
    ),
    {
        "dbt": dbt_cli_resource.configured(
            {
                "project_dir": DBT_PROJECT_PATH,
                "profiles_dir": DBT_PROFILES,
                # "vars": {"incremental_load_date": f"{today}"},
            },
        )
    },
)

4 Views

Open in Slack

Previous Next