Merge pull request #19 from datacoves/DCV-3760

[DCV-3760][DCV-3765] Profiles.yml refresh + partial parse fallback

Author: sconley-datacoves

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "dbt-core-interface"
-version = "1.1.6"
+version = "1.1.7"
 dynamic = []
 description = "Dbt Core Interface"
 authors = [

src/dbt_core_interface/project.py

@@ -375,9 +375,9 @@ def args(self, value: DbtConfiguration | dict[str, t.Any]) -> None:  # pyright:
         """Set the args for the DbtProject instance and update runtime config."""
         if isinstance(value, dict):
             value = dc_replace(self._args, **value)
+        self._args = value
         set_from_args(value, None)  # pyright: ignore[reportArgumentType]
         self.parse_project(write_manifest=True, reparse_configuration=True)
-        self._args = value
 
     def set_args(self, **kwargs: t.Any) -> None:
         """Set the args for the DbtProject instance."""
@@ -519,6 +519,20 @@ def parse_project(
     ) -> None:
         """Parse the dbt project and load manifest."""
         if reparse_configuration:
+            current = Path(self._args.profiles_dir).resolve()
+            standard_dirs = {
+                self.project_root.resolve(),
+                (Path.home() / ".dbt").resolve(),
+            }
+            env_dir = os.environ.get("DBT_PROFILES_DIR")
+            if env_dir:
+                standard_dirs.add(Path(env_dir).expanduser().resolve())
+            if current in standard_dirs:
+                self._args = dc_replace(
+                    self._args,
+                    profiles_dir=_get_profiles_dir(self.project_root),
+                )
+            set_from_args(self._args, None)  # pyright: ignore[reportArgumentType]
             self.runtime_config = RuntimeConfig.from_args(self._args)
             self.__manifest_loader = ManifestLoader(
                 self.runtime_config,
@@ -532,10 +546,28 @@ def parse_project(
             self.__manifest_loader.manifest.state_check = (
                 self.__manifest_loader.build_manifest_state_check()
             )
-            self._manifest = self.__manifest_loader.saved_manifest = self.__manifest_loader.load()
-            if not self.__manifest_loader.skip_parsing:
-                self._manifest.build_flat_graph()
-                self._manifest.build_group_map()
+            try:
+                self._manifest = self.__manifest_loader.saved_manifest = (
+                    self.__manifest_loader.load()
+                )
+                if not self.__manifest_loader.skip_parsing:
+                    self._manifest.build_flat_graph()
+                    self._manifest.build_group_map()
+            except Exception:
+                if self.__manifest_loader.saved_manifest is not None:
+                    logger.warning("Partial parse failed, forcing full reparse")
+                    self.__manifest_loader = ManifestLoader(
+                        self.runtime_config,
+                        self.runtime_config.load_dependencies(),
+                    )
+                    self._manifest = self.__manifest_loader.saved_manifest = (
+                        self.__manifest_loader.load()
+                    )
+                    if not self.__manifest_loader.skip_parsing:
+                        self._manifest.build_flat_graph()
+                        self._manifest.build_group_map()
+                else:
+                    raise
 
             self._sql_parser = None
             self._macro_parser = None

src/dbt_core_interface/server.py

@@ -38,6 +38,7 @@
     DbtConfiguration,
     DbtProject,
     ExecutionResult,
+    _get_profiles_dir,
 )
 from dbt_core_interface.watcher import DbtProjectWatcher
 
@@ -136,10 +137,26 @@ def _load_saved_state(runners: DbtProjectContainer) -> None:
             project_name = t.cast(str, entry.get("project"))
             if not project_name:
                 continue
+            project_dir = entry.get("project_dir")
+            saved_profiles_dir = entry.get("profiles_dir")
+            if saved_profiles_dir:
+                saved_path = Path(saved_profiles_dir).resolve()
+                standard_dirs: set[Path] = {(Path.home() / ".dbt").resolve()}
+                if project_dir:
+                    standard_dirs.add(Path(project_dir).resolve())
+                env_dir = os.environ.get("DBT_PROFILES_DIR")
+                if env_dir:
+                    standard_dirs.add(Path(env_dir).expanduser().resolve())
+                if saved_path in standard_dirs:
+                    profiles_dir = _get_profiles_dir(project_dir)
+                else:
+                    profiles_dir = saved_profiles_dir
+            else:
+                profiles_dir = _get_profiles_dir(project_dir)
             kwargs: dict[str, t.Any] = {
                 "target": entry.get("target"),
-                "profiles_dir": entry.get("profiles_dir"),
-                "project_dir": entry.get("project_dir"),
+                "project_dir": project_dir,
+                "profiles_dir": profiles_dir,
                 "threads": entry.get("threads", 1),
                 "vars": entry.get("vars", {}),
             }

src/dbt_core_interface/watcher.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import logging
+import os
 import threading
 import typing as t
 import weakref
@@ -115,6 +116,7 @@ def _monitor_loop(self) -> None:
             try:
                 change_level = self._check_for_changes()
                 if change_level:
+                    logger.info(f"Reparsing project (change_level={change_level})")
                     self._project.parse_project(
                         write_manifest=True, reparse_configuration=change_level > 1
                     )
@@ -123,6 +125,16 @@ def _monitor_loop(self) -> None:
 
             _ = self._stop_event.wait(self.check_interval)
 
+    def _all_profiles_yml_paths(self) -> list[Path]:
+        """Return all potential profiles.yml paths in priority order."""
+        paths = []
+        env_dir = os.environ.get("DBT_PROFILES_DIR")
+        if env_dir:
+            paths.append(Path(env_dir).expanduser().resolve() / "profiles.yml")
+        paths.append(self._project_or_raise.project_root / "profiles.yml")
+        paths.append(Path.home() / ".dbt" / "profiles.yml")
+        return paths
+
     def _initialize_file_mtimes(self) -> None:
         """Initialize the file modification time tracking."""
         for f_proxy in self._project_or_raise.manifest.files.values():
@@ -132,9 +144,8 @@ def _initialize_file_mtimes(self) -> None:
         self._mtimes[self._project_or_raise.dbt_project_yml] = (
             self._project_or_raise.dbt_project_yml.stat().st_mtime
         )
-        self._mtimes[self._project_or_raise.profiles_yml] = (
-            self._project_or_raise.profiles_yml.stat().st_mtime
-        )
+        for path in self._all_profiles_yml_paths():
+            self._mtimes[path] = path.stat().st_mtime if path.exists() else 0.0
         logger.debug(f"Initialized tracking for {len(self._mtimes)} files")
 
     def _check_for_changes(self) -> int:
@@ -143,12 +154,14 @@ def _check_for_changes(self) -> int:
         A return value of 0 means no changes, 1 means files were added/removed, and 2 means
         a configuration file was modified (dbt_project.yml or profiles.yml).
         """
-        for path in (self._project_or_raise.dbt_project_yml, self._project_or_raise.profiles_yml):
+        config_paths = [self._project_or_raise.dbt_project_yml, *self._all_profiles_yml_paths()]
+        for path in config_paths:
             try:
                 current_mtime = path.stat().st_mtime if path.exists() else 0.0
                 stamped_mtime = self._mtimes.get(path)
-                if stamped_mtime is None or current_mtime > stamped_mtime:
+                if stamped_mtime is None or current_mtime != stamped_mtime:
                     self._mtimes[path] = current_mtime
+                    logger.info(f"Config change detected: {path}")
                     return 2
             except OSError as e:
                 logger.warning(f"Error checking file {path}: {e}")