Allow multiple callbacks

cosmos/operators/local.py

@@ -160,7 +160,7 @@ def __init__(
         invocation_mode: InvocationMode | None = None,
         install_deps: bool = True,
         copy_dbt_packages: bool = settings.default_copy_dbt_packages,
-        callback: Callable[[str], None] | None = None,
+        callback: Callable[[str], None] | list[Callable[[str], None]] | None = None,
         callback_args: dict[str, Any] | None = None,
         should_store_compiled_sql: bool = True,
         should_upload_compiled_sql: bool = False,
@@ -508,7 +508,11 @@ def _handle_post_execution(self, tmp_project_dir: str, context: Context) -> None
             self._upload_sql_files(tmp_project_dir, "compiled")
         if self.callback:
             self.callback_args.update({"context": context})
-            self.callback(tmp_project_dir, **self.callback_args)
+            if isinstance(self.callback, list):
+                for callback_fn in self.callback:
+                    callback_fn(tmp_project_dir, **self.callback_args)
+            else:
+                self.callback(tmp_project_dir, **self.callback_args)
 
     def _handle_async_execution(self, tmp_project_dir: str, context: Context, async_context: dict[str, Any]) -> None:
         if async_context.get("teardown_task") and settings.enable_teardown_async_task:

tests/operators/test_local.py

@@ -1562,3 +1562,26 @@ def test_test_clone_project(create_symlinks_mock, copy_dbt_packages_mock, caplog
     assert f"Cloning project to writable temp directory {tmp_dir_path} from {project_dir}" in caplog.text
     assert "Copying dbt packages to temporary folder." in caplog.text
     assert "Completed copying dbt packages to temporary folder." in caplog.text
+
+
+@patch("cosmos.operators.local.AbstractDbtLocalBase.store_freshness_json")
+@patch("cosmos.operators.local.AbstractDbtLocalBase.store_compiled_sql")
+@patch("cosmos.operators.local.AbstractDbtLocalBase._override_rtif")
+def test_handle_post_execution_with_multiple_callbacks(
+    mock_override_rtif, mock_store_compiled_sql, mock_store_freshness_json
+):
+
+    multiple_callbacks = [MagicMock(), MagicMock(), MagicMock()]
+    operator = ConcreteDbtLocalBaseOperator(
+        profile_config=profile_config,
+        task_id="my-task",
+        project_dir="my/dir",
+        callback=multiple_callbacks,
+        callback_args={"arg1": "value1"},
+    )
+
+    context = {"dag_run": MagicMock(), "task": MagicMock()}
+    operator._handle_post_execution("/tmp/project_dir", context)
+
+    for callback_fn in multiple_callbacks:
+        callback_fn.assert_called_once_with("/tmp/project_dir", arg1="value1", context=context)