add support for full_refresh in template_fields

Author: Marthe Van Den Hende

.pre-commit-config.yaml

@@ -8,7 +8,7 @@ repos:
         name: Ensure end of file newline
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.1.1
+    rev: v1.15.0
     hooks:
       - id: mypy
         name: Static type checking

airflow_dbt_python/operators/dbt.py

@@ -399,7 +399,9 @@ class DbtRunOperator(DbtBaseOperator):
     https://docs.getdbt.com/reference/commands/run.
     """
 
-    template_fields = base_template_fields + selection_template_fields
+    template_fields = (
+        base_template_fields + selection_template_fields + ["full_refresh"]
+    )
 
     def __init__(
         self,
@@ -431,7 +433,9 @@ class DbtSeedOperator(DbtBaseOperator):
     https://docs.getdbt.com/reference/commands/seed.
     """
 
-    template_fields = base_template_fields + selection_template_fields
+    template_fields = (
+        base_template_fields + selection_template_fields + ["full_refresh"]
+    )
 
     def __init__(
         self,
@@ -500,7 +504,9 @@ class DbtCompileOperator(_GraphRunnableOperator):
     https://docs.getdbt.com/reference/commands/compile.
     """
 
-    template_fields = base_template_fields + selection_template_fields
+    template_fields = (
+        base_template_fields + selection_template_fields + ["full_refresh"]
+    )
 
     def __init__(
         self,
@@ -765,7 +771,9 @@ class DbtBuildOperator(DbtBaseOperator):
     https://docs.getdbt.com/reference/commands/build.
     """
 
-    template_fields = base_template_fields + selection_template_fields
+    template_fields = (
+        base_template_fields + selection_template_fields + ["full_refresh"]
+    )
 
     def __init__(
         self,

airflow_dbt_python/utils/configs.py

@@ -618,6 +618,8 @@ class TableMutabilityConfig(SelectionConfig):
     def __post_init__(self):
         """Call superclass __post_init__."""
         super().__post_init__()
+        if isinstance(self.full_refresh, str):
+            self.full_refresh = self.full_refresh.lower() == "true"
 
 
 @dataclasses.dataclass

tests/operators/test_dbt_build.py

@@ -234,3 +234,24 @@ def test_dbt_build_models_with_project_from_s3(
     build_result = execution_results["results"][0]
 
     assert build_result["status"] == RunStatus.Success
+
+
+def test_full_refresh_in_template_fields():
+    """Test that full_refresh is included in DbtBuildOperator template_fields."""
+    assert "full_refresh" in DbtBuildOperator.template_fields
+
+
+def test_full_refresh_templated():
+    """Test that full_refresh can be templated with Jinja."""
+    import pendulum
+    from airflow.models.dag import DAG
+
+    dag = DAG(dag_id="test_dag", start_date=pendulum.datetime(2025, 1, 1))
+    op = DbtBuildOperator(
+        task_id="dbt_task",
+        full_refresh="{{ params.full_refresh }}",
+        dag=dag,
+    )
+    context = {"params": {"full_refresh": "True"}}
+    op.render_template_fields(context)
+    assert op.full_refresh == "True"

tests/operators/test_dbt_compile.py

@@ -206,3 +206,24 @@ def test_dbt_compile_uses_correct_argument_according_to_version():
     assert getattr(op, "models", None) is None
     assert op.selector == "a-selector"
     assert getattr(op, "selector_name", None) is None
+
+
+def test_full_refresh_in_template_fields():
+    """Test that full_refresh is included in DbtCompileOperator template_fields."""
+    assert "full_refresh" in DbtCompileOperator.template_fields
+
+
+def test_full_refresh_templated():
+    """Test that full_refresh can be templated with Jinja."""
+    import pendulum
+    from airflow.models.dag import DAG
+
+    dag = DAG(dag_id="test_dag", start_date=pendulum.datetime(2025, 1, 1))
+    op = DbtCompileOperator(
+        task_id="dbt_task",
+        full_refresh="{{ params.full_refresh }}",
+        dag=dag,
+    )
+    context = {"params": {"full_refresh": "True"}}
+    op.render_template_fields(context)
+    assert op.full_refresh == "True"

tests/operators/test_dbt_run.py

@@ -428,3 +428,23 @@ def test_dbt_run_with_airflow_connection_and_profile(
         assert run_result["status"] == RunStatus.Success
         assert op.profiles_dir == profiles_file.parent
         assert execution_results["args"]["target"] == conn_id or target
+
+
+def test_full_refresh_in_template_fields():
+    """Test that full_refresh is included in DbtRunOperator template_fields."""
+    assert "full_refresh" in DbtRunOperator.template_fields
+
+
+def test_full_refresh_templated():
+    """Test that full_refresh can be templated with Jinja."""
+    from airflow.models.dag import DAG
+
+    dag = DAG(dag_id="test_dag", start_date=pendulum.datetime(2025, 1, 1))
+    op = DbtRunOperator(
+        task_id="dbt_task",
+        full_refresh="{{ params.full_refresh }}",
+        dag=dag,
+    )
+    context = {"params": {"full_refresh": "True"}}
+    op.render_template_fields(context)
+    assert op.full_refresh == "True"

tests/operators/test_dbt_seed.py

@@ -327,3 +327,24 @@ def test_dbt_seed_with_airflow_connection_and_profile(
         assert run_result["status"] == RunStatus.Success
         assert op.profiles_dir == profiles_file.parent
         assert execution_results["args"]["target"] == conn_id or target
+
+
+def test_full_refresh_in_template_fields():
+    """Test that full_refresh is included in DbtSeedOperator template_fields."""
+    assert "full_refresh" in DbtSeedOperator.template_fields
+
+
+def test_full_refresh_templated():
+    """Test that full_refresh can be templated with Jinja."""
+    import pendulum
+    from airflow.models.dag import DAG
+
+    dag = DAG(dag_id="test_dag", start_date=pendulum.datetime(2025, 1, 1))
+    op = DbtSeedOperator(
+        task_id="dbt_task",
+        full_refresh="{{ params.full_refresh }}",
+        dag=dag,
+    )
+    context = {"params": {"full_refresh": "True"}}
+    op.render_template_fields(context)
+    assert op.full_refresh == "True"

tests/utils/test_configs.py

@@ -506,3 +506,36 @@ def test_base_config_does_not_override_when_value_in_environment(
         )
     assert config.fail_fast is None
     assert config.require_generic_test_arguments_property is None
+
+
+def test_table_mutability_config_full_refresh_string_coercion(
+    profiles_file, dbt_project_file
+):
+    """Test that full_refresh string values are coerced to booleans."""
+    config = RunTaskConfig(
+        profiles_dir=profiles_file.parent,
+        project_dir=dbt_project_file.parent,
+        full_refresh="True",
+    )
+    assert config.full_refresh is True
+
+    config = RunTaskConfig(
+        profiles_dir=profiles_file.parent,
+        project_dir=dbt_project_file.parent,
+        full_refresh="False",
+    )
+    assert config.full_refresh is False
+
+    config = RunTaskConfig(
+        profiles_dir=profiles_file.parent,
+        project_dir=dbt_project_file.parent,
+        full_refresh=None,
+    )
+    assert config.full_refresh is None
+
+    config = RunTaskConfig(
+        profiles_dir=profiles_file.parent,
+        project_dir=dbt_project_file.parent,
+        full_refresh=True,
+    )
+    assert config.full_refresh is True