WIP

Author: z3z1ma

src/dbt_core_interface/project.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-# pyright: reportDeprecated=false,reportPrivateImportUsage=false,reportAny=false,reportUnknownMemberType=false,reportUnknownVariableType=false
+# pyright: reportDeprecated=false,reportPrivateImportUsage=false,reportAny=false,reportUnknownMemberType=false,reportUnknownVariableType=false,reportUnnecessaryComparison=false
 """Minimal dbt-core interface for in-memory manifest management and SQL execution."""
 
 from __future__ import annotations
@@ -476,6 +476,305 @@ def get_node_by_path(self, path: Path | str) -> ManifestNode | None:
         for node in self.manifest.nodes.values():
             if self.project_root / node.original_file_path == path:
                 return node
+        return None
+
+    def update_node_from_content(  # noqa: C901
+        self, path_: Path | str, /, content: str | None = None, force_update: bool = False
+    ) -> bool:
+        """Update a specific node in the manifest from file content or disk."""
+        path = Path(path_)
+        if not path.is_absolute():
+            path = self.project_root / path
+        path = path.resolve()
+
+        if not self._is_dbt_file(path):
+            return False
+
+        if content is None:
+            try:
+                content = path.read_text(encoding="utf-8")
+            except (OSError, UnicodeDecodeError) as e:
+                logger.warning(f"Failed to read file {path}: {e}")
+                return False
+
+        existing_node = self.get_node_by_path(path)
+
+        if not force_update and existing_node and existing_node.raw_code == content:
+            return True
+
+        try:
+            with self._manifest_lock:
+                if existing_node:
+                    unique_id = existing_node.unique_id
+
+                    _ = self.manifest.nodes.pop(unique_id, None)
+                    _ = self.manifest.macros.pop(unique_id, None)
+
+                if self._is_macro_file(path):
+                    processed_nodes = self._parse_macro_content(path, content)
+                else:
+                    processed_nodes = self._parse_model_content(path, content)
+
+                for node in processed_nodes:
+                    if node.resource_type == NodeType.Macro:
+                        self.manifest.macros[node.unique_id] = node
+                    else:
+                        self.manifest.nodes[node.unique_id] = node
+
+                    if hasattr(node, "depends_on"):
+                        process_node(self.runtime_config, self.manifest, node)
+
+                if existing_node and existing_node.raw_code in self.__compilation_cache:
+                    del self.__compilation_cache[existing_node.raw_code]
+
+                logger.debug(f"Successfully updated node(s) from {path}")
+                return True
+
+        except Exception as e:
+            logger.error(f"Failed to update node from {path}: {e}")
+            if existing_node:
+                if existing_node.resource_type == NodeType.Macro:
+                    self.manifest.macros[existing_node.unique_id] = existing_node
+                else:
+                    self.manifest.nodes[existing_node.unique_id] = existing_node
+            return False
+
+    def update_nodes_by_paths(self, *paths: Path | str) -> dict[str, bool]:
+        """Update multiple nodes from their disk files."""
+        results: dict[str, bool] = {}
+        for path in paths:
+            results[str(path)] = self.update_node_from_content(path)
+        return results
+
+    def detect_new_files(self) -> list[Path]:
+        """Detect new dbt files that aren't in the manifest."""
+        known_files = set()
+
+        for node in self.manifest.nodes.values():
+            known_files.add(self.project_root / node.original_file_path)
+        for macro in self.manifest.macros.values():
+            known_files.add(self.project_root / macro.original_file_path)
+
+        discovered = []
+        for pattern in ["**/*.sql", "**/*.py"]:
+            for path in self.project_root.glob(pattern):
+                if path.is_file() and self._is_dbt_file(path):
+                    if path not in known_files:
+                        discovered.append(path)
+
+        return discovered
+
+    def add_new_files(self, *paths: Path | str) -> dict[str, bool]:
+        """Add new files to the manifest."""
+        if not paths:
+            paths = tuple(self.detect_new_files())
+
+        results = {}
+        for path in paths:
+            try:
+                success = self.update_node_from_content(path, force_update=True)
+                results[str(path)] = success
+                if success:
+                    logger.info(f"Added new file to manifest: {path}")
+            except Exception as e:
+                logger.error(f"Failed to add new file {path}: {e}")
+                results[str(path)] = False
+
+        return results
+
+    def create_project_watcher(self, check_interval: float = 2.0) -> DbtProjectWatcher:
+        """Create a project watcher for automatic incremental updates."""
+        return DbtProjectWatcher(self, check_interval)
+
+    def _is_dbt_file(self, path: Path) -> bool:  # noqa: C901
+        """Check if a file is a valid dbt file using runtime config paths."""
+        if not path.is_file():
+            return False
+
+        parts = path.parts
+        if any(skip in parts for skip in ["target", "logs", "dbt_packages", ".git", "__pycache__"]):
+            return False
+
+        if path.suffix.lower() not in [".sql", ".py"]:
+            return False
+
+        try:
+            relative_path = path.relative_to(self.project_root)
+            path_str = str(relative_path)
+
+            for model_path in self.runtime_config.model_paths:
+                if path_str.startswith(model_path):
+                    return True
+
+            for macro_path in self.runtime_config.macro_paths:
+                if path_str.startswith(macro_path):
+                    return True
+
+            for snapshot_path in self.runtime_config.snapshot_paths:
+                if path_str.startswith(snapshot_path):
+                    return True
+
+        except ValueError:
+            return False
+
+        return False
+
+    def _is_macro_file(self, path: Path) -> bool:
+        """Check if a file is a macro file using runtime config."""
+        try:
+            relative_path = str(path.relative_to(self.project_root))
+            return any(
+                relative_path.startswith(macro_path)
+                for macro_path in self.runtime_config.macro_paths
+            )
+        except ValueError:
+            return False
+
+    def _parse_macro_content(self, path: Path, content: str) -> list[ManifestNode]:
+        """Parse macro content and return list of macro nodes."""
+        parser = self.macro_parser
+        nodes = []
+
+        try:
+            parsed_macros = parser.parse_remote(content)
+            for macro in parsed_macros:
+                nodes.append(macro)
+        except Exception as e:
+            logger.error(f"Failed to parse macro content from {path}: {e}")
+            raise
+
+        return nodes
+
+    def _parse_model_content(self, path: Path, content: str) -> list[ManifestNode]:
+        """Parse model content and return list of model nodes."""
+        relative_path = path.relative_to(self.project_root)
+
+        parser = self.sql_parser
+        nodes = []
+
+        try:
+            node = parser.parse_remote(content, str(relative_path))
+            nodes.append(node)
+        except Exception as e:
+            logger.error(f"Failed to parse model content from {path}: {e}")
+            raise
+
+        return nodes
+
+
+@t.final
+class DbtProjectWatcher:
+    """Watch dbt files for changes and automatically update the manifest."""
+
+    def __init__(self, project: DbtProject, check_interval: float = 2.0):
+        self.project = project
+        self.check_interval = check_interval
+        self._mtimes: dict[Path, float] = {}
+        self._running = False
+        self._thread: threading.Thread | None = None
+        self._stop_event = threading.Event()
+
+    def start(self) -> None:
+        """Start monitoring files for changes."""
+        if self._running:
+            return
+
+        self._running = True
+        self._stop_event.clear()
+        self._thread = threading.Thread(target=self._monitor_loop, daemon=True)
+        self._thread.start()
+        logger.info("Project watcher started")
+
+    def stop(self) -> None:
+        """Stop monitoring files."""
+        if not self._running:
+            return
+
+        self._running = False
+        self._stop_event.set()
+        if self._thread:
+            self._thread.join(timeout=5.0)
+        logger.info("Project watcher stopped")
+
+    def _monitor_loop(self) -> None:
+        """Run the main monitoring loop."""
+        self._initialize_file_mtimes()
+
+        while self._running and not self._stop_event.is_set():
+            try:
+                self._check_for_changes()
+                self._check_for_new_files()
+            except Exception as e:
+                logger.error(f"Error in project watcher loop: {e}")
+
+            _ = self._stop_event.wait(self.check_interval)
+
+    def _initialize_file_mtimes(self) -> None:
+        """Initialize the file modification time tracking."""
+        try:
+            for node in self.project.manifest.nodes.values():
+                path = self.project.project_root / node.original_file_path
+                if path.exists():
+                    self._mtimes[path] = path.stat().st_mtime
+
+            for macro in self.project.manifest.macros.values():
+                path = self.project.project_root / macro.original_file_path
+                if path.exists():
+                    self._mtimes[path] = path.stat().st_mtime
+
+            logger.debug(f"Initialized tracking for {len(self._mtimes)} files")
+        except Exception as e:
+            logger.error(f"Failed to initialize file mtimes: {e}")
+
+    def _check_for_changes(self) -> None:
+        """Check for changes in tracked files."""
+        edits: list[Path] = []
+
+        for path, stamped_mtime in list(self._mtimes.items()):
+            try:
+                if not path.exists():
+                    _ = self._mtimes.pop(path, None)
+                    logger.info(f"File deleted: {path}")
+                    continue
+
+                current_mtime = path.stat().st_mtime
+                if current_mtime > stamped_mtime:
+                    edits.append(path)
+                    self._mtimes[path] = current_mtime
+
+            except OSError as e:
+                logger.warning(f"Error checking file {path}: {e}")
+                continue
+
+        if edits:
+            logger.info(f"Detected changes in {len(edits)} files")
+            results = self.project.update_nodes_by_paths(*edits)
+
+            for path, success in results.items():
+                if success:
+                    logger.info(f"Successfully updated: {path}")
+                else:
+                    logger.error(f"Failed to update: {path}")
+
+    def _check_for_new_files(self) -> None:
+        """Check for new dbt files."""
+        try:
+            added_files = self.project.detect_new_files()
+            if added_files:
+                logger.info(f"Detected {len(added_files)} new files")
+                results = self.project.add_new_files(*added_files)
+
+                for path, success in results.items():
+                    if success:
+                        path = Path(path)
+                        if path.exists():
+                            self._mtimes[path] = path.stat().st_mtime
+                        logger.info(f"Successfully added new file: {path}")
+                    else:
+                        logger.error(f"Failed to add new file: {path}")
+
+        except Exception as e:
+            logger.error(f"Error checking for new files: {e}")
 
 
 # Import protection for optional dependencies