chore: constrain all dbt import surface area to project py

Author: z3z1ma

src/dbt_core_interface/dbt_templater/templater.py

@@ -9,7 +9,6 @@
 from contextlib import suppress
 from functools import cached_property
 
-from dbt_common.exceptions import CompilationError as DbtCompilationException
 from sqlfluff.core.errors import SQLFluffSkipFile, SQLTemplaterError
 from sqlfluff.core.templaters.base import (
     TemplatedFile,
@@ -56,6 +55,8 @@ def process(  # pyright: ignore[reportIncompatibleMethodOverride]
         **kwargs: t.Any,
     ) -> tuple[TemplatedFile | None, list[SQLTemplaterError]]:
         """Compile a dbt model and return the compiled SQL."""
+        from dbt_common.exceptions import CompilationError as DbtCompilationException
+
         fname_absolute_path = (
             os.path.abspath(fname) if fname not in ("stdin", "<string input>") else fname
         )

src/dbt_core_interface/project.py

@@ -46,7 +46,7 @@ def _patched_adapter_accessor(config: t.Any) -> t.Any:
 from dbt.contracts.graph.nodes import ManifestNode, SourceDefinition
 from dbt.flags import set_from_args
 from dbt.parser.manifest import ManifestLoader, process_node
-from dbt.parser.read_files import FileDiff, InputFile
+from dbt.parser.read_files import FileDiff, InputFile, ReadFilesFromFileSystem
 from dbt.parser.sql import SqlBlockParser, SqlMacroParser
 from dbt.task.sql import SqlCompileRunner
 from dbt.tracking import disable_tracking
@@ -62,7 +62,13 @@ def _patched_adapter_accessor(config: t.Any) -> t.Any:
 
 
 disable_tracking()
-set_invocation_context(get_env())
+
+
+def _set_invocation_context() -> None:
+    set_invocation_context(get_env())
+
+
+_set_invocation_context()
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@@ -267,6 +273,8 @@ def __repr__(self) -> str:  # pyright: ignore[reportImplicitOverride]
         """Return a string representation of the DbtProject instance."""
         return f"DbtProject(name={self.project_name}, root={self.project_root}, last_parsed_at={self._last_parsed_at})"
 
+    set_invocation_context = staticmethod(_set_invocation_context)
+
     @classmethod
     def from_config(cls, config: DbtConfiguration) -> DbtProject:
         """Create project from configuration."""
@@ -417,6 +425,14 @@ def create_adapter(
 
         return self._adapter
 
+    def create_reader(self) -> ReadFilesFromFileSystem:
+        """Create a file reader for the project."""
+        return ReadFilesFromFileSystem(
+            all_projects=self.runtime_config.load_dependencies(),
+            files={},
+            saved_files=self.manifest.files,
+        )
+
     def parse_project(
         self, write_manifest: bool = False, reparse_configuration: bool = False
     ) -> None:

src/dbt_core_interface/watcher.py

@@ -8,10 +8,6 @@
 import typing as t
 from pathlib import Path
 
-from dbt.parser.read_files import ReadFilesFromFileSystem
-from dbt_common.clients.system import get_env
-from dbt_common.context import set_invocation_context
-
 if t.TYPE_CHECKING:
     from dbt_core_interface.project import DbtProject
 
@@ -48,11 +44,7 @@ def __init__(
         self._project = project
         self.check_interval = check_interval
 
-        self.reader = ReadFilesFromFileSystem(
-            all_projects=self._project.runtime_config.load_dependencies(),
-            files={},
-            saved_files=self._project.manifest.files,
-        )
+        self.reader = project.create_reader()
 
         self._mtimes: dict[Path, float] = {}
         self._running = False
@@ -87,7 +79,7 @@ def stop(self) -> None:
     def _monitor_loop(self) -> None:
         """Run the main monitoring loop."""
         self._initialize_file_mtimes()
-        set_invocation_context(get_env())
+        self._project.set_invocation_context()
 
         while self._running and not self._stop_event.is_set():
             try: