feat: add --mode flag to create_external_models command

Adds a --mode option to create_external_models with three values:
- overwrite (default): current behavior, replaces same-gateway entries
- sync: preserves hand-edited metadata, only syncs columns from DB,
  warns on stale entries but keeps them
- sync_prune: same as sync but removes stale entries

Introduces CreateExternalModelsMode enum following the codebase's
(str, Enum) convention with is_* properties and a default classproperty.

Fixes a bug where external models loaded from the existing YAML file
were always re-fetched from the database, preventing stale detection.
The new logic only fetches FQNs referenced by non-external model
depends_on, excluding known internal models to prevent conversion.

Signed-off-by: Christoph Meier <chrism216@gmail.com>

Author: christophmeier90

sqlmesh/cli/main.py

@@ -977,6 +977,14 @@ def rollback(obj: Context) -> None:
     is_flag=True,
     help="Raise an error if the external model is missing in the database",
 )
+@click.option(
+    "--mode",
+    type=click.Choice(["overwrite", "sync", "sync_prune"], case_sensitive=False),
+    default="overwrite",
+    help="The mode for updating external models. overwrite replaces all entries (default), "
+    "sync syncs columns while preserving metadata and warns on stale entries, "
+    "sync_prune also removes stale entries.",
+)
 @click.pass_obj
 @error_handler
 @cli_analytics

sqlmesh/core/context.py

@@ -92,7 +92,10 @@
 from sqlmesh.core.plan.definition import UserProvidedFlags
 from sqlmesh.core.reference import ReferenceGraph
 from sqlmesh.core.scheduler import Scheduler, CompletionStatus
-from sqlmesh.core.schema_loader import create_external_models_file
+from sqlmesh.core.schema_loader import (
+    CreateExternalModelsMode,
+    create_external_models_file,
+)
 from sqlmesh.core.selector import Selector, NativeSelector
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
@@ -2486,15 +2489,24 @@ def rollback(self) -> None:
         self._new_state_sync().rollback()
 
     @python_api_analytics
-    def create_external_models(self, strict: bool = False) -> None:
+    def create_external_models(
+        self,
+        strict: bool = False,
+        mode: t.Union[CreateExternalModelsMode, str] = CreateExternalModelsMode.default,
+    ) -> None:
         """Create a file to document the schema of external models.
 
         The external models file contains all columns and types of external models, allowing for more
         robust lineage, validation, and optimizations.
 
         Args:
             strict: If True, raise an error if the external model is missing in the database.
+            mode: The mode for updating external models. "overwrite" replaces all entries (default),
+                "sync" syncs columns while preserving metadata and warns on stale entries,
+                "sync_prune" also removes stale entries.
         """
+        if isinstance(mode, str):
+            mode = CreateExternalModelsMode(mode)
         if not self._models:
             self.load(update_schemas=False)
 
@@ -2529,6 +2541,7 @@ def create_external_models(self, strict: bool = False) -> None:
                 max_workers=self.concurrent_tasks,
                 strict=strict,
                 all_models=self._models,
+                mode=mode,
             )
 
     @python_api_analytics

sqlmesh/core/schema_loader.py

@@ -2,6 +2,7 @@
 
 import typing as t
 from concurrent.futures import ThreadPoolExecutor
+from enum import Enum
 from pathlib import Path
 
 from sqlglot import exp
@@ -11,10 +12,34 @@
 from sqlmesh.core.engine_adapter import EngineAdapter
 from sqlmesh.core.model.definition import Model
 from sqlmesh.core.state_sync import StateReader
-from sqlmesh.utils import UniqueKeyDict, yaml
+from sqlmesh.utils import UniqueKeyDict, classproperty, yaml
 from sqlmesh.utils.errors import SQLMeshError
 
 
+class CreateExternalModelsMode(str, Enum):
+    """Determines how existing entries are handled when creating external models."""
+
+    OVERWRITE = "overwrite"
+    SYNC = "sync"
+    SYNC_PRUNE = "sync_prune"
+
+    @classproperty
+    def default(cls) -> CreateExternalModelsMode:
+        return CreateExternalModelsMode.OVERWRITE
+
+    @property
+    def is_overwrite(self) -> bool:
+        return self == CreateExternalModelsMode.OVERWRITE
+
+    @property
+    def is_sync(self) -> bool:
+        return self == CreateExternalModelsMode.SYNC
+
+    @property
+    def is_sync_prune(self) -> bool:
+        return self == CreateExternalModelsMode.SYNC_PRUNE
+
+
 def create_external_models_file(
     path: Path,
     models: UniqueKeyDict[str, Model],
@@ -25,6 +50,7 @@ def create_external_models_file(
     max_workers: int = 1,
     strict: bool = False,
     all_models: t.Optional[t.Dict[str, Model]] = None,
+    mode: CreateExternalModelsMode = CreateExternalModelsMode.default,
 ) -> None:
     """Create or replace a YAML file with column and types of all columns in all external models.
 
@@ -40,16 +66,18 @@ def create_external_models_file(
         all_models: FQN to model across all loaded repos. When provided, a dependency is only
             classified as external if it is absent from this full set. This prevents cross-repo
             internal models from being misclassified as external in multi-repo setups.
+        mode: The mode for updating external models. "overwrite" replaces all entries (default),
+            "update" syncs columns while preserving metadata and warns on stale entries,
+            "update_prune" also removes stale entries.
     """
     known_models: t.Dict[str, Model] = all_models if all_models is not None else models
-    external_model_fqns = set()
 
-    for fqn, model in models.items():
-        if model.kind.is_external:
-            external_model_fqns.add(fqn)
-        for dep in model.depends_on:
-            if dep not in known_models:
-                external_model_fqns.add(dep)
+    external_model_fqns = {
+        dep
+        for model in models.values()
+        for dep in model.depends_on
+        if dep not in known_models or known_models[dep].kind.is_external
+    }
 
     # Make sure we don't convert internal models into external ones.
     existing_model_fqns = state_reader.nodes_exist(external_model_fqns, exclude_external=True)
@@ -79,15 +107,78 @@ def create_external_models_file(
             if columns
         ]
 
-        # dont clobber existing entries from other gateways
-        entries_to_keep = (
-            [e for e in yaml.load(path) if e.get("gateway", None) != gateway]
-            if path.exists()
-            else []
+        if mode.is_overwrite:
+            # dont clobber existing entries from other gateways
+            entries_to_keep = (
+                [e for e in yaml.load(path) if e.get("gateway", None) != gateway]
+                if path.exists()
+                else []
+            )
+
+            with open(path, "w", encoding="utf-8") as file:
+                yaml.dump(entries_to_keep + schemas, file)
+        else:
+            _write_external_models_file_update(path, schemas, gateway, mode=mode)
+
+
+def _write_external_models_file_update(
+    path: Path,
+    schemas: t.List[t.Dict[str, t.Any]],
+    gateway: t.Optional[str],
+    mode: CreateExternalModelsMode,
+) -> None:
+    """Write external models file for sync and sync_prune modes.
+
+    Preserves hand-edited metadata on existing entries, only syncing columns from the DB.
+    sync mode warns about stale entries but keeps them.
+    sync_prune mode removes stale entries.
+    """
+    existing_entries = yaml.load(path) if path.exists() else []
+
+    other_gateway_entries: t.List[t.Dict[str, t.Any]] = []
+    same_gateway_entries: t.List[t.Dict[str, t.Any]] = []
+
+    for entry in existing_entries:
+        if entry.get("gateway", None) != gateway:
+            other_gateway_entries.append(entry)
+        else:
+            same_gateway_entries.append(entry)
+
+    same_gateway_by_name: t.Dict[str, t.Dict[str, t.Any]] = {}
+    for entry in same_gateway_entries:
+        same_gateway_by_name[entry["name"]] = entry
+
+    result = list(other_gateway_entries)
+    matched_in_schemas: t.Set[str] = {s["name"] for s in schemas}
+
+    for new_schema in schemas:
+        name = new_schema["name"]
+        if name in same_gateway_by_name:
+            existing = same_gateway_by_name[name]
+            existing["columns"] = new_schema["columns"]
+            result.append(existing)
+        else:
+            result.append(new_schema)
+
+    stale_count = 0
+    for entry in same_gateway_entries:
+        name = entry["name"]
+        if name not in matched_in_schemas:
+            stale_count += 1
+            if mode.is_sync:
+                get_console().log_warning(
+                    f"External model '{name}' is no longer referenced but was preserved. "
+                    "Use --mode sync_prune to automatically remove stale entries."
+                )
+                result.append(entry)
+
+    if stale_count and mode.is_sync_prune:
+        get_console().log_warning(
+            f"Removed {stale_count} stale external model(s) that are no longer referenced."
         )
 
-        with open(path, "w", encoding="utf-8") as file:
-            yaml.dump(entries_to_keep + schemas, file)
+    with open(path, "w", encoding="utf-8") as file:
+        yaml.dump(result, file)
 
 
 def get_columns(

sqlmesh/magics.py

@@ -738,12 +738,20 @@ def migrate(self, context: Context, line: str) -> None:
         action="store_true",
         help="Raise an error if the external model is missing in the database",
     )
+    @argument(
+        "--mode",
+        choices=["overwrite", "sync", "sync_prune"],
+        default="overwrite",
+        help="The mode for updating external models. overwrite replaces all entries (default), "
+        "sync syncs columns while preserving metadata and warns on stale entries, "
+        "sync_prune also removes stale entries.",
+    )
     @line_magic
     @pass_sqlmesh_context
     def create_external_models(self, context: Context, line: str) -> None:
         """Create a schema file containing external model schemas."""
         args = parse_argstring(self.create_external_models, line)
-        context.create_external_models(strict=args.strict)
+        context.create_external_models(strict=args.strict, mode=args.mode)
 
     @magic_arguments()
     @argument(