diff --git a/python_modules/libraries/dagster-dbt/dagster_dbt/core/dbt_event_iterator.py b/python_modules/libraries/dagster-dbt/dagster_dbt/core/dbt_event_iterator.py
index 90cbe50a1598b..e6eafb13f8129 100644
--- a/python_modules/libraries/dagster-dbt/dagster_dbt/core/dbt_event_iterator.py
+++ b/python_modules/libraries/dagster-dbt/dagster_dbt/core/dbt_event_iterator.py
@@ -277,15 +277,15 @@ def _map_fn(event: DbtDagsterEventType) -> DbtDagsterEventType:
         # opening multiple connections to the same database when a write connection, such
         # as the one dbt uses, is open.
         event_stream = self
-        try:
+        if (
+            self._dbt_cli_invocation.adapter
+            and self._dbt_cli_invocation.adapter.__class__.__name__ == "DuckDBAdapter"
+        ):
             from dbt.adapters.duckdb import DuckDBAdapter
 
             if isinstance(self._dbt_cli_invocation.adapter, DuckDBAdapter):
                 event_stream = exhaust_iterator_and_yield_results_with_exception(self)
 
-        except ImportError:
-            pass
-
         def _threadpool_wrap_map_fn() -> (
             Iterator[Union[Output, AssetMaterialization, AssetObservation, AssetCheckResult]]
         ):
diff --git a/python_modules/libraries/dagster-dbt/dagster_dbt/core/resource.py b/python_modules/libraries/dagster-dbt/dagster_dbt/core/resource.py
index 827ba6c6facd9..c19e44274a1ca 100644
--- a/python_modules/libraries/dagster-dbt/dagster_dbt/core/resource.py
+++ b/python_modules/libraries/dagster-dbt/dagster_dbt/core/resource.py
@@ -379,7 +379,8 @@ def _initialize_adapter(self, cli_vars) -> BaseAdapter:
 
         # If the dbt adapter is DuckDB, set the access mode to READ_ONLY, since DuckDB only allows
         # simultaneous connections for read-only access.
-        try:
+
+        if config.credentials and config.credentials.__class__.__name__ == "DuckDBCredentials":
             from dbt.adapters.duckdb.credentials import DuckDBCredentials
 
             if isinstance(config.credentials, DuckDBCredentials):
@@ -391,12 +392,6 @@ def _initialize_adapter(self, cli_vars) -> BaseAdapter:
                 with pushd(self.project_dir):
                     config.credentials.path = os.fspath(Path(config.credentials.path).absolute())
 
-        except ImportError:
-            logger.warning(
-                "An error was encountered when creating a handle to the dbt adapter in Dagster.",
-                exc_info=True,
-            )
-
         cleanup_event_logger()
 
         # reset adapters list in case we have instantiated an adapter before in this process