Remove the fallback operator for async task

cosmos/operators/_asynchronous/base.py

@@ -30,9 +30,8 @@ def _create_async_operator_class(profile_type: str, dbt_class: str) -> Any:
         module = importlib.import_module(module_path)
         operator_class = getattr(module, class_name)
         return operator_class
-    except (ModuleNotFoundError, AttributeError):
-        log.info("Error in loading class: %s. falling back to DbtRunLocalOperator", class_path)
-        return DbtRunLocalOperator
+    except (ModuleNotFoundError, AttributeError) as e:
+        raise ImportError(f"Error in loading class: {class_path}. Unable to find the specified operator class.") from e
 
 
 class DbtRunAirflowAsyncFactoryOperator(DbtRunLocalOperator):  # type: ignore[misc]

tests/operators/_asynchronous/test_base.py

@@ -8,25 +8,33 @@
 from cosmos.operators._asynchronous import TeardownAsyncOperator
 from cosmos.operators._asynchronous.base import DbtRunAirflowAsyncFactoryOperator, _create_async_operator_class
 from cosmos.operators._asynchronous.bigquery import DbtRunAirflowAsyncBigqueryOperator
+from cosmos.operators._asynchronous.databricks import DbtRunAirflowAsyncDatabricksOperator
 from cosmos.operators.local import DbtRunLocalOperator
 
+_ASYNC_PROFILE = ["bigquery", "databricks"]
+
 
 @pytest.mark.parametrize(
     "profile_type, dbt_class, expected_operator_class",
     [
         ("bigquery", "DbtRun", DbtRunAirflowAsyncBigqueryOperator),
-        ("snowflake", "DbtRun", DbtRunLocalOperator),
-        ("bigquery", "DbtTest", DbtRunLocalOperator),
+        ("databricks", "DbtRun", DbtRunAirflowAsyncDatabricksOperator),
     ],
 )
-def test_create_async_operator_class_success(profile_type, dbt_class, expected_operator_class):
+def test_create_async_operator_class(profile_type, dbt_class, expected_operator_class):
     """Test the successful loading of the async operator class."""
 
     operator_class = _create_async_operator_class(profile_type, dbt_class)
 
     assert operator_class == expected_operator_class
 
 
+def test_create_async_operator_class_unsupported():
+
+    with pytest.raises(ImportError, match="Error in loading class"):
+        _create_async_operator_class("test_profile", "DbtRun")
+
+
 @pytest.fixture
 def profile_config_mock():
     """Fixture to create a mock ProfileConfig."""
@@ -46,13 +54,6 @@ def test_create_async_operator_class_valid():
         assert result == mock_class
 
 
-def test_create_async_operator_class_fallback():
-    """Test _create_async_operator_class falls back to DbtRunLocalOperator when import fails."""
-    with patch("cosmos.operators._asynchronous.base.importlib.import_module", side_effect=ModuleNotFoundError):
-        result = _create_async_operator_class("bigquery", "DbtRun")
-        assert result == DbtRunLocalOperator
-
-
 class MockAsyncOperator(DbtRunLocalOperator):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)