hi guys. I want to ask, how to do dbt cli command like "$ dbt run --full-refresh --select my_incremental_model+" in dagster as a job so we can run it to full refresh the "my_incremental_model" ???

If you build your own op that uses the dbt CLI resource, you can pass any command line argument you want to

dbt run

, including the full refresh flag.

context.resources.dbt.run(
            select=context.op_config["select"],
            exclude=context.op_config["exclude"],
            vars=context.op_config["vars"],
            full_refresh=context.op_config["full_refresh"],
        )

@owen is there a way to do this with software-defined assets?

yep! the

dbt_cli_resource

can be configured with

{..., "full_refresh": True}

@owen is `{..., "full_refresh": True}`means that all the dbt models will be full_refresh? i just want a few of incremental model can be full_refresh or not.. is there a way to config it? the default full_refresh False and sometimes we can config it in launchpad UI to set it to True to do full refresh. Thanks in advance

as far as I know, dbt doesn't support targeting individual models for full refresh other than by just running a subset of your models in a given run in terms of making a particular run execute with a configurable full-refresh flag, you can do that by making the

dbt_cli_resource

configurable:

@configured(dbt_cli_resource, config_schema={"full_refresh": Field(bool, default_value=False)})
def custom_dbt_cli_resource(config):
    return {
        "full_refresh": config["full_refresh"],
        "project-dir": DBT_PROJECT_DIR,
        "profiles-dir": DBT_PROFILES_DIR
    }

from there, in the UI, you can kick off a run with extra config by shift+clicking the "materialize" button (so you'd select the models you want to full refresh, then shift+click the button). then you can add config to that run that would look like

resources:
  dbt:
    config:
      full_refresh: True

One more question. What is the Field in "Field(bool, default_value=False)"? Is it a func? How to define it?

you can import it from the dagster module (from dagster import Field)

Wow I really appreciate your help @owen