revert to use get_or_load_models for nonsqlmodels and dbt

Author: themisvaltinos

sqlmesh/core/loader.py

@@ -68,6 +68,13 @@ class LoadedProject:
 
 
 class CacheBase(abc.ABC):
+    @abc.abstractmethod
+    def get_or_load_models(
+        self, target_path: Path, loader: t.Callable[[], t.List[Model]]
+    ) -> t.List[Model]:
+        """Get or load all models from cache."""
+        pass
+
     @abc.abstractmethod
     def put(self, models: t.List[Model], path: Path) -> bool:
         pass
@@ -291,31 +298,30 @@ def _load_external_models(
         if external_models_path.exists() and external_models_path.is_dir():
             paths_to_load.extend(self._glob_paths(external_models_path, extension=".yaml"))
 
+        def _load(path: Path) -> t.List[Model]:
+            try:
+                with open(path, "r", encoding="utf-8") as file:
+                    return [
+                        create_external_model(
+                            defaults=self.config.model_defaults.dict(),
+                            path=path,
+                            project=self.config.project,
+                            audit_definitions=audits,
+                            **{
+                                "dialect": self.config.model_defaults.dialect,
+                                "default_catalog": self.context.default_catalog,
+                                **row,
+                            },
+                        )
+                        for row in YAML().load(file.read())
+                    ]
+            except Exception as ex:
+                raise ConfigError(self._failed_to_load_model_error(path, ex))
+
         for path in paths_to_load:
             self._track_file(path)
-            external_models = cache.get(path)
 
-            if not external_models:
-                try:
-                    with open(path, "r", encoding="utf-8") as file:
-                        external_models = [
-                            create_external_model(
-                                defaults=self.config.model_defaults.dict(),
-                                path=path,
-                                project=self.config.project,
-                                audit_definitions=audits,
-                                **{
-                                    "dialect": self.config.model_defaults.dialect,
-                                    "default_catalog": self.context.default_catalog,
-                                    **row,
-                                },
-                            )
-                            for row in YAML().load(file.read())
-                        ]
-
-                    cache.put(external_models, path)
-                except Exception as ex:
-                    raise ConfigError(f"Failed to load model definition at '{path}'.\n{ex}")
+            external_models = cache.get_or_load_models(path, lambda: _load(path))
 
             # external models with no explicit gateway defined form the base set
             for model in external_models:
@@ -843,6 +849,20 @@ def __init__(self, loader: SqlMeshLoader, config_path: Path):
             self.config_path = config_path
             self._model_cache = ModelCache(self.config_path / c.CACHE)
 
+        def get_or_load_models(
+            self, target_path: Path, loader: t.Callable[[], t.List[Model]]
+        ) -> t.List[Model]:
+            models = self._model_cache.get_or_load(
+                self._cache_entry_name(target_path),
+                self._model_cache_entry_id(target_path),
+                loader=loader,
+            )
+
+            for model in models:
+                model._path = target_path
+
+            return models
+
         def put(self, models: t.List[Model], path: Path) -> bool:
             return self._model_cache.put(
                 models,

sqlmesh/core/model/cache.py

@@ -10,7 +10,7 @@
 from sqlglot.schema import MappingSchema
 
 from sqlmesh.core import constants as c
-from sqlmesh.core.model.definition import ExternalModel, Model, SqlModel
+from sqlmesh.core.model.definition import ExternalModel, Model, SqlModel, _Model
 from sqlmesh.utils.cache import FileCache
 from sqlmesh.utils.hashing import crc32
 from sqlmesh.utils.process import PoolExecutor, create_process_pool_executor
@@ -40,6 +40,30 @@ def __init__(self, path: Path):
             prefix="model_definition",
         )
 
+    def get_or_load(
+        self, name: str, entry_id: str = "", *, loader: t.Callable[[], t.List[Model]]
+    ) -> t.List[Model]:
+        """Returns an existing cached model definition or loads and caches a new one.
+        Args:
+            name: The name of the entry.
+            entry_id: The unique entry identifier. Used for cache invalidation.
+            loader: Used to load a new model definition when no cached instance was found.
+        Returns:
+            The model definition.
+        """
+        cache_entry = self._file_cache.get(name, entry_id)
+        if isinstance(cache_entry, list) and isinstance(seq_get(cache_entry, 0), _Model):
+            return cache_entry
+
+        models = loader()
+        if isinstance(models, list) and isinstance(seq_get(models, 0), (SqlModel, ExternalModel)):
+            # make sure we preload full_depends_on
+            for model in models:
+                model.full_depends_on
+
+            self._file_cache.put(name, entry_id, value=models)
+        return models
+
     def put(self, models: t.List[Model], name: str, entry_id: str = "") -> bool:
         if isinstance(models, list) and isinstance(seq_get(models, 0), (SqlModel, ExternalModel)):
             # make sure we preload full_depends_on

sqlmesh/dbt/loader.py

@@ -147,13 +147,9 @@ def _to_sqlmesh(config: BMC, context: DbtContext) -> Model:
                         )
                         continue
 
-                    cached_models = cache.get(model.path)
-
-                    if cached_models:
-                        sqlmesh_model = cached_models[0]
-                    else:
-                        sqlmesh_model = _to_sqlmesh(model, context)
-                        cache.put([sqlmesh_model], model.path)
+                    sqlmesh_model = cache.get_or_load_models(
+                        model.path, loader=lambda: [_to_sqlmesh(model, context)]
+                    )[0]
 
                     models[sqlmesh_model.fqn] = sqlmesh_model
 
@@ -346,6 +342,19 @@ def __init__(
             cache_path = loader.config_path / c.CACHE / target.name
             self._model_cache = ModelCache(cache_path)
 
+        def get_or_load_models(
+            self, target_path: Path, loader: t.Callable[[], t.List[Model]]
+        ) -> t.List[Model]:
+            models = self._model_cache.get_or_load(
+                self._cache_entry_name(target_path),
+                self._cache_entry_id(target_path),
+                loader=loader,
+            )
+            for model in models:
+                model._path = target_path
+
+            return models
+
         def put(self, models: t.List[Model], path: Path) -> bool:
             return self._model_cache.put(
                 models,