Store compiled_sql even when task fails (fixes issue #369) (#671)

Update `DbtLocalBaseOperator` code to store `compiled_sql` prior to
exception handling so that when a task fails, the `compiled_sql` can still
be reviewed.

In the process found and fixed a related bug where `compiled_sql` was
being dropped on some operations due to the way that the `full_refresh`
field was being added to the `template_fields`.

Closes #369

Fixes bug introduced in
#623 where
compiled_sql was being lost in `DbtSeedLocalOperator` and
`DbtRunLocalOperator`

Co-authored-by: Andrew Greenburg <[email protected]>
(cherry picked from commit ee91ece)

cosmos/operators/local.py

@@ -245,8 +245,8 @@ def run_command(
                     logger.info("Outlets: %s", outlets)
                     self.register_dataset(inlets, outlets)
 
-                self.exception_handling(result)
                 self.store_compiled_sql(tmp_project_dir, context)
+                self.exception_handling(result)
                 if self.callback:
                     self.callback(tmp_project_dir)
 
@@ -395,6 +395,7 @@ class DbtSeedLocalOperator(DbtLocalBaseOperator):
     """
 
     ui_color = "#F58D7E"
+    template_fields: Sequence[str] = DbtLocalBaseOperator.template_fields + ("full_refresh",)  # type: ignore[operator]
 
     def __init__(self, full_refresh: bool = False, **kwargs: Any) -> None:
         self.full_refresh = full_refresh
@@ -433,6 +434,7 @@ class DbtRunLocalOperator(DbtLocalBaseOperator):
 
     ui_color = "#7352BA"
     ui_fgcolor = "#F4F2FC"
+    template_fields: Sequence[str] = DbtLocalBaseOperator.template_fields + ("full_refresh",)  # type: ignore[operator]
 
     def __init__(self, full_refresh: bool = False, **kwargs: Any) -> None:
         self.full_refresh = full_refresh

tests/operators/test_local.py

@@ -419,8 +419,8 @@ def test_calculate_openlineage_events_completes_openlineage_errors(mock_processo
 @pytest.mark.parametrize(
     "operator_class,expected_template",
     [
-        (DbtSeedLocalOperator, ("env", "vars", "full_refresh")),
-        (DbtRunLocalOperator, ("env", "vars", "full_refresh")),
+        (DbtSeedLocalOperator, ("env", "vars", "compiled_sql", "full_refresh")),
+        (DbtRunLocalOperator, ("env", "vars", "compiled_sql", "full_refresh")),
     ],
 )
 def test_dbt_base_operator_template_fields(operator_class, expected_template):