Make dbt auto-materialize policy rules configurable

[darijavz]

Hi, is there a way to add an auto-materialize rules for dbt models in meta config? I would like something like that:

models:
  my_models:
    marts:
      +meta:
          dagster:
            auto_materialize_policy:
              type: eager
              skip_on_all_parents_not_updated:
                require_update_for_all_parent_partitions: False

However, as I understand currently it is only possible to configure skip_on_all_parents_not_updated rules using DagsterDbtTranslator for all assets, which is not desired in my case. Any help is appreciated.

[rexledesma]

meta config doesn't support all the possible ways to configure an auto-materialization policy with auto-materialize rules.

We recommend that you use the DagsterDbtTranslator instead: https://docs.dagster.io/integrations/dbt/reference#customizing-auto-materialize-policies.

[darijavz]

But this way it will override how auto-materialize policy works for all dbt assets. Is there a workaround if I only want to set the require_update_for_all_parent_partitions=False for one specific asset?

[rexledesma]

You could just select for that single model in the code.

class CustomDagsterDbtTranslator(DagsterDbtTranslator):
    def get_auto_materialize_policy(
        self, dbt_resource_props: Mapping[str, Any]
    ) -> Optional[AutoMaterializePolicy]:
        auto_materialize_policy = super().get_auto_materialize_policy(dbt_resource_props)
        
        model_names_without_rules = ...
        if auto_materialize_policy and dbt_resource_props["name"] in model_names_without_rules:
            return auto_materialize_policy.without_rules(...)
            
        return auto_materialize_policy

[OwenKephart]

The same answer applies for setting a custom automation condition (of any sort) for a specific asset:

class CustomDagsterDbtTranslator(DagsterDbtTranslator):
    def get_automation_condition(
        self, dbt_resource_props: Mapping[str, Any]
    ) -> Optional[AutomationCondition]:

        model_names_with_specific_condition = ...
        if dbt_resource_props["name"] in model_names_with_specific_condition:
            return AutomationCondition.something_specific()
        else:
            return AutomationCondition.another_condition()