Fix: Allow create_external_models to run for projects with star projections and support dbt formatted projects. (#880)

Author: crericha

sqlmesh/cli/main.py

@@ -36,11 +36,15 @@ def cli(
     if ctx.invoked_subcommand == "version":
         return
 
+    load = True
+
     if len(paths) == 1:
         path = os.path.abspath(paths[0])
         if ctx.invoked_subcommand == "init":
             ctx.obj = path
             return
+        elif ctx.invoked_subcommand == "create_external_models":
+            load = False
 
     # Delegates the execution of the --help option to the corresponding subcommand
     if "--help" in sys.argv:
@@ -53,9 +57,10 @@ def cli(
         paths=paths,
         config=config,
         gateway=gateway,
+        load=load,
     )
 
-    if not context.models:
+    if load and not context.models:
         raise click.ClickException(
             f"`{paths}` doesn't seem to have any models... cd into the proper directory or specify the path(s) with -p."
         )

sqlmesh/core/context.py

@@ -847,12 +847,13 @@ def create_external_models(self) -> None:
         The schema file contains all columns and types of external models, allowing for more robust
         lineage, validation, and optimizations.
         """
+        models = self._models or self._loader.load(self, update_schemas=False).models
         for path, config in self.configs.items():
             create_schema_file(
                 path=path / c.SCHEMA_YAML,
                 models={
                     name: model
-                    for name, model in self._models.items()
+                    for name, model in models.items()
                     if self.config_for_model(model) is config
                 },
                 adapter=self._engine_adapter,

sqlmesh/core/loader.py

@@ -69,7 +69,9 @@ def update_model_schemas(dag: DAG[str], models: UniqueKeyDict[str, Model]) -> No
         if external:
             if "*" in model.columns_to_types:
                 raise ConfigError(
-                    f"Can't expand SELECT * expression for model '{name}'. Projections for models that use external sources must be specified explicitly at '{model._path}'"
+                    f"Can't expand SELECT * expression for model '{name}'."
+                    " Projections for models that use external sources must be specified explicitly"
+                    " or use external models (https://sqlmesh.readthedocs.io/en/stable/concepts/models/external_models)."
                 )
         elif model.mapping_schema:
             try:
@@ -97,7 +99,7 @@ def __init__(self) -> None:
         self._path_mtimes: t.Dict[Path, float] = {}
         self._dag: DAG[str] = DAG()
 
-    def load(self, context: Context) -> LoadedProject:
+    def load(self, context: Context, update_schemas: bool = True) -> LoadedProject:
         """
         Loads all hooks, macros, and models in the context's path
 
@@ -128,7 +130,8 @@ def load(self, context: Context) -> LoadedProject:
         models = self._load_models(macros, hooks, jinja_macros)
         for model in models.values():
             self._add_model_to_dag(model)
-        update_model_schemas(self._dag, models)
+        if update_schemas:
+            update_model_schemas(self._dag, models)
 
         audits = self._load_audits()
 
@@ -169,6 +172,24 @@ def _load_models(
     def _load_audits(self) -> UniqueKeyDict[str, Audit]:
         """Loads all audits."""
 
+    def _load_external_models(self) -> UniqueKeyDict[str, Model]:
+        models: UniqueKeyDict = UniqueKeyDict("models")
+        for context_path, config in self._context.configs.items():
+            path = Path(context_path / c.SCHEMA_YAML)
+
+            if path.exists():
+                self._track_file(path)
+
+                with open(path, "r", encoding="utf-8") as file:
+                    for row in YAML().load(file.read()):
+                        model = create_external_model(
+                            **row,
+                            dialect=config.model_defaults.dialect,
+                            path=path,
+                        )
+                        models[model.name] = model
+        return models
+
     def _add_model_to_dag(self, model: Model) -> None:
         self._dag.graph[model.name] = set()
         self._dag.add(model.name, model.depends_on)
@@ -240,24 +261,6 @@ def _load_models(
 
         return models
 
-    def _load_external_models(self) -> UniqueKeyDict[str, Model]:
-        models: UniqueKeyDict = UniqueKeyDict("models")
-        for context_path, config in self._context.configs.items():
-            path = Path(context_path / c.SCHEMA_YAML)
-
-            if path.exists():
-                self._track_file(path)
-
-                with open(path, "r", encoding="utf-8") as file:
-                    for row in YAML().load(file.read()):
-                        model = create_external_model(
-                            **row,
-                            dialect=config.model_defaults.dialect,
-                            path=path,
-                        )
-                        models[model.name] = model
-        return models
-
     def _load_sql_models(
         self, macros: MacroRegistry, hooks: HookRegistry, jinja_macros: JinjaMacroRegistry
     ) -> UniqueKeyDict[str, Model]:

sqlmesh/dbt/loader.py

@@ -108,6 +108,8 @@ def _load_models(
                 }
             )
 
+        models.update(self._load_external_models())
+
         return models
 
     def _load_audits(self) -> UniqueKeyDict[str, Audit]: