Repository: TobikoData/sqlmesh
Pull request #2776 (2024-06-14 20:06:16)

Feat: Support 'enabled' flag in model definitions
Now that we support project-wide variables in model meta, it also makes sense to support the `enabled` flag to allow users to conditionally enable / disable certain models.


Patch:
diff --git a/docs/concepts/models/overview.md b/docs/concepts/models/overview.md
index 040b73d3..f29f7fda 100644
--- a/docs/concepts/models/overview.md
+++ b/docs/concepts/models/overview.md
@@ -302,6 +302,9 @@ MODEL (
 ### allow_partials
 - Indicates that this model can be executed for partial (incomplete) data intervals. By default, each model processes only complete intervals to prevent common mistakes caused by partial data. The size of the interval is determined by the model's [interval_unit](#interval_unit). Setting `allow_partials` to `true` overrides this behavior, indicating that the model may process a segment of input data that is missing some of the data points. Please note that setting this attribute to `true` results in the disregard of the [cron](#cron) attribute.
 
+### enabled
+- Whether the model is enabled. This attribute is `true` by default. Setting it to `false` causes SQLMesh to ignore this model when loading the project.
+
 ## Incremental Model Properties
 
 For models that are incremental, the following parameters can be specified in the `kind`'s definition.
diff --git a/docs/reference/model_configuration.md b/docs/reference/model_configuration.md
index aeecbf40..f50d9d5e 100644
--- a/docs/reference/model_configuration.md
+++ b/docs/reference/model_configuration.md
@@ -8,31 +8,32 @@ Learn more about specifying SQLMesh model properties in the [model concepts over
 
 Configuration options for SQLMesh model properties. Supported by all model kinds other than [`SEED` models](#seed-models).
 
-| Option                | Description                                                                                                                                                                                                                                                                                      |       Type        | Required |
-| --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | :---------------: | :------: |
-| `name`                | The model name. Must include at least a qualifying schema (`<schema>.<model>`) and may include a catalog (`<catalog>.<schema>.<model>`). Can be omitted if [infer_names](#model-naming) is set to true.                                                                                                                                                        |        str        |    N     |
-| `kind`                | The model kind ([Additional Details](#model-kind-properties)). (Default: `VIEW`)                                                                                                                                                                                                                 |    str \| dict    |    N     |
-| `audits`              | SQLMesh [audits](../concepts/audits.md) that should run against the model's output                                                                                                                                                                                                               |    array[str]     |    N     |
-| `dialect`             | The SQL dialect in which the model's query is written. All SQL dialects [supported by the SQLGlot library](https://github.com/tobymao/sqlglot/blob/main/sqlglot/dialects/dialect.py) are allowed.                                                                                                |        str        |    N     |
-| `owner`               | The owner of a model; may be used for notification purposes                                                                                                                                                                                                                                      |        str        |    N     |
-| `stamp`               | Arbitrary string used to indicate a model's version without changing the model name                                                                                                                                                                                                              |        str        |    N     |
-| `tags`                | Arbitrary strings used to organize or classify a model                                                                                                                                                                                                                                           |    array[str]     |    N     |
-| `cron`                | The cron expression specifying how often the model should be refreshed. (Default: `@daily`)                                                                                                                                                                                                      |        str        |    N     |
-| `interval_unit`       | The temporal granularity of the model's data intervals. Supported values: `year`, `month`, `day`, `hour`, `half_hour`, `quarter_hour`, `five_minute`. (Default: inferred from `cron`)                                                                                                            |        str        |    N     |
-| `start`               | The date/time that determines the earliest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`. (Default: `yesterday`)                                                                          |    str \| int     |    N     |
-| `end`                 | The date/time that determines the latest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`.                                                                                                   |    str \| int     |    N     |
-| `grains`              | The column(s) whose combination uniquely identifies each row in the model                                                                                                                                                                                                                        | str \| array[str] |    N     |
-| `references`          | The model column(s) used to join to other models' grains                                                                                                                                                                                                                                         | str \| array[str] |    N     |
-| `depends_on`          | Models on which this model depends. (Default: dependencies inferred from model code)                                                                                                                                                                                                             |    array[str]     |    N     |
-| `storage_format`      | The storage format that should be used to store physical tables; only applicable to engines such as Spark                                                                                                                                                                                        |        str        |    N     |
-| `partitioned_by`      | The column(s) and/or column expressions used define a model's partitioning key. Required for the `INCREMENTAL_BY_PARTITION` model kind. Optional for all other model kinds; used to partition the model's physical table in engines that support partitioning. | str \| array[str] |    N     |
-| `clustered_by`        | The column(s) used to cluster the model's physical table; only applicable to engines that support clustering                                                                                                                                                                                     |        str        |    N     |
-| `columns`             | The column names and data types returned by the model. Disables [automatic inference of column names and types](../concepts/models/overview.md#conventions) from the SQL query.                                                                                                                  |    array[str]     |    N     |
-| `physical_properties` | A key-value mapping of arbitrary properties specific to the target engine that are applied to the model table / view in the physical layer. Specified as key-value pairs (`key = value`).                                                                                                        |       dict        |    N     |
-| `virtual_properties`  | A key-value mapping of arbitrary properties specific to the target engine that are applied to the model view in the virtual layer. Specified as key-value pairs (`key = value`).                                                                                                                 |       dict        |    N     |
-| `allow_partials`      | Whether this model can process partial (incomplete) data intervals                                                                                                                                                                                                                               |       bool        |    N     |
-| `description`         | Description of the model. Automatically registered in the SQL engine's table COMMENT field or equivalent (if supported by the engine).                                                                                                                                                           |        str        |    N     |
-| `column_descriptions` | A key-value mapping of column names to column comments that will be registered in the SQL engine's table COMMENT field (if supported by the engine). Specified as key-value pairs (`column_name = 'column comment'`). If present, [inline column comments](../concepts/models/overview.md#inline-column-comments) will not be registered in the SQL engine.                                                                                                        |       dict        |    N     |
+| Option                | Description                                                                                                                                                                                                                                                                                                                                                 | Type              | Required |
+|-----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:-----------------:|:--------:|
+| `name`                | The model name. Must include at least a qualifying schema (`<schema>.<model>`) and may include a catalog (`<catalog>.<schema>.<model>`). Can be omitted if [infer_names](#model-naming) is set to true.                                                                                                                                                     | str               | N        |
+| `kind`                | The model kind ([Additional Details](#model-kind-properties)). (Default: `VIEW`)                                                                                                                                                                                                                                                                            | str \| dict       | N        |
+| `audits`              | SQLMesh [audits](../concepts/audits.md) that should run against the model's output                                                                                                                                                                                                                                                                          | array[str]        | N        |
+| `dialect`             | The SQL dialect in which the model's query is written. All SQL dialects [supported by the SQLGlot library](https://github.com/tobymao/sqlglot/blob/main/sqlglot/dialects/dialect.py) are allowed.                                                                                                                                                           | str               | N        |
+| `owner`               | The owner of a model; may be used for notification purposes                                                                                                                                                                                                                                                                                                 | str               | N        |
+| `stamp`               | Arbitrary string used to indicate a model's version without changing the model name                                                                                                                                                                                                                                                                         | str               | N        |
+| `tags`                | Arbitrary strings used to organize or classify a model                                                                                                                                                                                                                                                                                                      | array[str]        | N        |
+| `cron`                | The cron expression specifying how often the model should be refreshed. (Default: `@daily`)                                                                                                                                                                                                                                                                 | str               | N        |
+| `interval_unit`       | The temporal granularity of the model's data intervals. Supported values: `year`, `month`, `day`, `hour`, `half_hour`, `quarter_hour`, `five_minute`. (Default: inferred from `cron`)                                                                                                                                                                       | str               | N        |
+| `start`               | The date/time that determines the earliest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`. (Default: `yesterday`)                                                                                                                                     | str \| int        | N        |
+| `end`                 | The date/time that determines the latest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`.                                                                                                                                                              | str \| int        | N        |
+| `grains`              | The column(s) whose combination uniquely identifies each row in the model                                                                                                                                                                                                                                                                                   | str \| array[str] | N        |
+| `references`          | The model column(s) used to join to other models' grains                                                                                                                                                                                                                                                                                                    | str \| array[str] | N        |
+| `depends_on`          | Models on which this model depends. (Default: dependencies inferred from model code)                                                                                                                                                                                                                                                                        | array[str]        | N        |
+| `storage_format`      | The storage format that should be used to store physical tables; only applicable to engines such as Spark                                                                                                                                                                                                                                                   | str               | N        |
+| `partitioned_by`      | The column(s) and/or column expressions used define a model's partitioning key. Required for the `INCREMENTAL_BY_PARTITION` model kind. Optional for all other model kinds; used to partition the model's physical table in engines that support partitioning.                                                                                              | str \| array[str] | N        |
+| `clustered_by`        | The column(s) used to cluster the model's physical table; only applicable to engines that support clustering                                                                                                                                                                                                                                                | str               | N        |
+| `columns`             | The column names and data types returned by the model. Disables [automatic inference of column names and types](../concepts/models/overview.md#conventions) from the SQL query.                                                                                                                                                                             | array[str]        | N        |
+| `physical_properties` | A key-value mapping of arbitrary properties specific to the target engine that are applied to the model table / view in the physical layer. Specified as key-value pairs (`key = value`).                                                                                                                                                                   | dict              | N        |
+| `virtual_properties`  | A key-value mapping of arbitrary properties specific to the target engine that are applied to the model view in the virtual layer. Specified as key-value pairs (`key = value`).                                                                                                                                                                            | dict              | N        |
+| `allow_partials`      | Whether this model can process partial (incomplete) data intervals                                                                                                                                                                                                                                                                                          | bool              | N        |
+| `description`         | Description of the model. Automatically registered in the SQL engine's table COMMENT field or equivalent (if supported by the engine).                                                                                                                                                                                                                      | str               | N        |
+| `column_descriptions` | A key-value mapping of column names to column comments that will be registered in the SQL engine's table COMMENT field (if supported by the engine). Specified as key-value pairs (`column_name = 'column comment'`). If present, [inline column comments](../concepts/models/overview.md#inline-column-comments) will not be registered in the SQL engine. | dict              | N        |
+| `enabled`             | Whether the model is enabled. This attribute is `true` by default. Setting it to `false` causes SQLMesh to ignore this model when loading the project.                                                                                                                                                                                                      | bool              | N        |
 
 ### Model defaults
 
diff --git a/examples/sushi/models/disabled.sql b/examples/sushi/models/disabled.sql
new file mode 100644
index 00000000..e97fcfbf
--- /dev/null
+++ b/examples/sushi/models/disabled.sql
@@ -0,0 +1,6 @@
+MODEL (
+  name sushi.disabled,
+  enabled False,
+);
+
+SELECT 1 AS a;
diff --git a/sqlmesh/cli/example_project.py b/sqlmesh/cli/example_project.py
index 1f998b3d..990260aa 100644
--- a/sqlmesh/cli/example_project.py
+++ b/sqlmesh/cli/example_project.py
@@ -4,6 +4,7 @@ from pathlib import Path
 
 import click
 from sqlglot import Dialect
+from sqlmesh.utils.date import yesterday_ds
 
 
 class ProjectTemplate(Enum):
@@ -25,6 +26,7 @@ default_gateway: local
 
 model_defaults:
   dialect: {dialect}
+  start: {yesterday_ds()}
 """,
         ProjectTemplate.AIRFLOW: f"""gateways:
   local:
@@ -42,6 +44,7 @@ default_scheduler:
 
 model_defaults:
   dialect: {dialect}
+  start: {yesterday_ds()}
 """,
         ProjectTemplate.DBT: """from pathlib import Path
 
diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py
index 88d4111d..82fbd712 100644
--- a/sqlmesh/core/console.py
+++ b/sqlmesh/core/console.py
@@ -130,16 +130,32 @@ class Console(abc.ABC):
         """Stop the snapshot promotion progress."""
 
     @abc.abstractmethod
-    def start_migration_progress(self, total_tasks: int) -> None:
-        """Indicates that a new migration progress has begun."""
+    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
+        """Indicates that a new snapshot migration progress has begun."""
 
     @abc.abstractmethod
-    def update_migration_progress(self, num_tasks: int) -> None:
-        """Update the migration progress."""
+    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
+        """Update the snapshot migration progress."""
 
     @abc.abstractmethod
-    def stop_migration_progress(self, success: bool = True) -> None:
-        """Stop the migration progress."""
+    def log_migration_status(self, success: bool = True) -> None:
+        """Log the finished migration status."""
+
+    @abc.abstractmethod
+    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
+        """Stop the snapshot migration progress."""
+
+    @abc.abstractmethod
+    def start_env_migration_progress(self, total_tasks: int) -> None:
+        """Indicates that a new environment migration progress has begun."""
+
+    @abc.abstractmethod
+    def update_env_migration_progress(self, num_tasks: int) -> None:
+        """Update the environment migration progress."""
+
+    @abc.abstractmethod
+    def stop_env_migration_progress(self, success: bool = True) -> None:
+        """Stop the environment migration progress."""
 
     @abc.abstractmethod
     def show_model_difference_summary(
@@ -220,6 +236,19 @@ class Console(abc.ABC):
         """Show table summary diff."""
 
 
+def make_progress_bar(message: str, console: t.Optional[RichConsole] = None) -> Progress:
+    return Progress(
+        TextColumn(f"[bold blue]{message}", justify="right"),
+        BarColumn(bar_width=40),
+        "[progress.percentage]{task.percentage:>3.1f}%",
+        "•",
+        srich.BatchColumn(),
+        "•",
+        TimeElapsedColumn(),
+        console=console,
+    )
+
+
 class TerminalConsole(Console):
     """A rich based implementation of the console."""
 
@@ -252,6 +281,9 @@ class TerminalConsole(Console):
         self.migration_progress: t.Optional[Progress] = None
         self.migration_task: t.Optional[TaskID] = None
 
+        self.env_migration_progress: t.Optional[Progress] = None
+        self.env_migration_task: t.Optional[TaskID] = None
+
         self.loading_status: t.Dict[uuid.UUID, Status] = {}
 
         self.verbose = verbose
@@ -280,16 +312,7 @@ class TerminalConsole(Console):
     ) -> None:
         """Indicates that a new snapshot evaluation progress has begun."""
         if not self.evaluation_progress_live:
-            self.evaluation_total_progress = Progress(
-                TextColumn("[bold blue]Evaluating models", justify="right"),
-                BarColumn(bar_width=40),
-                "[progress.percentage]{task.percentage:>3.1f}%",
-                "•",
-                srich.BatchColumn(),
-                "•",
-                TimeElapsedColumn(),
-                console=self.console,
-            )
+            self.evaluation_total_progress = make_progress_bar("Evaluating models", self.console)
 
             self.evaluation_model_progress = Progress(
                 TextColumn("{task.fields[view_name]}", justify="right"),
@@ -372,16 +395,7 @@ class TerminalConsole(Console):
     ) -> None:
         """Indicates that a new creation progress has begun."""
         if self.creation_progress is None:
-            self.creation_progress = Progress(
-                TextColumn("[bold blue]Creating physical tables", justify="right"),
-                BarColumn(bar_width=40),
-                "[progress.percentage]{task.percentage:>3.1f}%",
-                "•",
-                srich.BatchColumn(),
-                "•",
-                TimeElapsedColumn(),
-                console=self.console,
-            )
+            self.creation_progress = make_progress_bar("Creating physical table", self.console)
 
             self.creation_progress.start()
             self.creation_task = self.creation_progress.add_task(
@@ -468,19 +482,10 @@ class TerminalConsole(Console):
         self.environment_naming_info = EnvironmentNamingInfo()
         self.default_catalog = None
 
-    def start_migration_progress(self, total_tasks: int) -> None:
-        """Indicates that a new migration progress has begun."""
+    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
+        """Indicates that a new snapshot migration progress has begun."""
         if self.migration_progress is None:
-            self.migration_progress = Progress(
-                TextColumn("[bold blue]Migrating snapshots", justify="right"),
-                BarColumn(bar_width=40),
-                "[progress.percentage]{task.percentage:>3.1f}%",
-                "•",
-                srich.BatchColumn(),
-                "•",
-                TimeElapsedColumn(),
-                console=self.console,
-            )
+            self.migration_progress = make_progress_bar("Migrating snapshots", self.console)
 
             self.migration_progress.start()
             self.migration_task = self.migration_progress.add_task(
@@ -488,19 +493,51 @@ class TerminalConsole(Console):
                 total=total_tasks,
             )
 
-    def update_migration_progress(self, num_tasks: int) -> None:
+    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
         """Update the migration progress."""
         if self.migration_progress is not None and self.migration_task is not None:
             self.migration_progress.update(self.migration_task, refresh=True, advance=num_tasks)
 
-    def stop_migration_progress(self, success: bool = True) -> None:
+    def log_migration_status(self, success: bool = True) -> None:
+        """Log the migration status."""
+        if self.migration_progress is not None:
+            self.migration_progress = None
+            if success:
+                self.log_success("The migration has been completed successfully")
+
+    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
         """Stop the migration progress."""
         self.migration_task = None
         if self.migration_progress is not None:
             self.migration_progress.stop()
-            self.migration_progress = None
             if success:
-                self.log_success("The migration has been completed successfully")
+                self.log_success("All snapshots have been migrated successfully")
+
+    def start_env_migration_progress(self, total_tasks: int) -> None:
+        """Indicates that a new environment migration has begun."""
+        if self.env_migration_progress is None:
+            self.env_migration_progress = make_progress_bar("Migrating environments", self.console)
+            self.env_migration_progress.start()
+            self.env_migration_task = self.env_migration_progress.add_task(
+                "Migrating environments...",
+                total=total_tasks,
+            )
+
+    def update_env_migration_progress(self, num_tasks: int) -> None:
+        """Update the environment migration progress."""
+        if self.env_migration_progress is not None and self.env_migration_task is not None:
+            self.env_migration_progress.update(
+                self.env_migration_task, refresh=True, advance=num_tasks
+            )
+
+    def stop_env_migration_progress(self, success: bool = True) -> None:
+        """Stop the environment migration progress."""
+        self.env_migration_task = None
+        if self.env_migration_progress is not None:
+            self.env_migration_progress.stop()
+            self.env_migration_progress = None
+            if success:
+                self.log_success("All environments have been migrated successfully")
 
     def show_model_difference_summary(
         self,
@@ -1717,12 +1754,12 @@ class DatabricksMagicConsole(CaptureTerminalConsole):
         self.promotion_status = (0, 0)
         print(f"Virtual Update {'succeeded' if success else 'failed'}")
 
-    def start_migration_progress(self, total_tasks: int) -> None:
+    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
         """Indicates that a new migration progress has begun."""
         self.migration_status = (0, total_tasks)
         print("Starting Migration")
 
-    def update_migration_progress(self, num_tasks: int) -> None:
+    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
         """Update the migration progress."""
         num_migrations, total_migrations = self.migration_status
         num_migrations += num_tasks
@@ -1730,10 +1767,32 @@ class DatabricksMagicConsole(CaptureTerminalConsole):
         if num_migrations % 5 == 0:
             print(f"Migration Updated {num_migrations}/{total_migrations}")
 
-    def stop_migration_progress(self, success: bool = True) -> None:
+    def log_migration_status(self, success: bool = True) -> None:
+        """Log the migration status."""
+        print(f"Migration {'succeeded' if success else 'failed'}")
+
+    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
         """Stop the migration progress."""
         self.migration_status = (0, 0)
-        print(f"Migration {'succeeded' if success else 'failed'}")
+        print(f"Snapshot migration {'succeeded' if success else 'failed'}")
+
+    def start_env_migration_progress(self, total_tasks: int) -> None:
+        """Indicates that a new migration progress has begun."""
+        self.env_migration_status = (0, total_tasks)
+        print("Starting Environment migration")
+
+    def update_env_migration_progress(self, num_tasks: int) -> None:
+        """Update the migration progress."""
+        num_migrations, total_migrations = self.env_migration_status
+        num_migrations += num_tasks
+        self.env_migration_status = (num_migrations, total_migrations)
+        if num_migrations % 5 == 0:
+            print(f"Environment migration Updated {num_migrations}/{total_migrations}")
+
+    def stop_env_migration_progress(self, success: bool = True) -> None:
+        """Stop the migration progress."""
+        self.env_migration_status = (0, 0)
+        print(f"Environment migration {'succeeded' if success else 'failed'}")
 
 
 class DebuggerTerminalConsole(TerminalConsole):
@@ -1808,14 +1867,26 @@ class DebuggerTerminalConsole(TerminalConsole):
     def stop_promotion_progress(self, success: bool = True) -> None:
         self._write(f"Stopping promotion with success={success}")
 
-    def start_migration_progress(self, total_tasks: int) -> None:
+    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
         self._write(f"Starting migration for {total_tasks} snapshots")
 
-    def update_migration_progress(self, num_tasks: int) -> None:
+    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
         self._write(f"Migration {num_tasks}")
 
-    def stop_migration_progress(self, success: bool = True) -> None:
-        self._write(f"Stopping migration with success={success}")
+    def log_migration_status(self, success: bool = True) -> None:
+        self._write(f"Migration finished with success={success}")
+
+    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
+        self._write(f"Stopping snapshot migration with success={success}")
+
+    def start_env_migration_progress(self, total_tasks: int) -> None:
+        self._write(f"Starting migration for {total_tasks} environments")
+
+    def update_env_migration_progress(self, num_tasks: int) -> None:
+        self._write(f"Environment migration {num_tasks}")
+
+    def stop_env_migration_progress(self, success: bool = True) -> None:
+        self._write(f"Stopping environment migration with success={success}")
 
     def show_model_difference_summary(
         self,
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 32552305..17aad875 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -412,6 +412,8 @@ class GenericContext(BaseContext, t.Generic[C]):
             A new instance of the updated or inserted model.
         """
         model = self.get_model(model, raise_if_missing=True)
+        if not model.enabled:
+            raise SQLMeshError(f"The disabled model '{model.name}' cannot be upserted")
         path = model._path
 
         # model.copy() can't be used here due to a cached state that can be a part of a model instance.
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index c85b9a14..24d6be36 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -328,7 +328,8 @@ class SqlMeshLoader(Loader):
                     )
 
                 model = cache.get_or_load_model(path, _load)
-                models[model.fqn] = model
+                if model.enabled:
+                    models[model.fqn] = model
 
                 if isinstance(model, SeedModel):
                     seed_path = model.seed_path
diff --git a/sqlmesh/core/model/common.py b/sqlmesh/core/model/common.py
index 1d0f04d8..ebb6e643 100644
--- a/sqlmesh/core/model/common.py
+++ b/sqlmesh/core/model/common.py
@@ -131,6 +131,7 @@ bool_validator = field_validator(
     "disable_restatement",
     "insert_overwrite",
     "allow_partials",
+    "enabled",
     mode="before",
     check_fields=False,
 )(parse_bool)
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 5a7084d4..9b2867a8 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -180,7 +180,7 @@ class _Model(ModelMeta, frozen=True):
                             value=exp.to_table(field_value, dialect=self.dialect),
                         )
                     )
-                elif field_name not in ("column_descriptions_", "default_catalog"):
+                elif field_name not in ("column_descriptions_", "default_catalog", "enabled"):
                     expressions.append(
                         exp.Property(
                             this=field_info.alias or field_name,
diff --git a/sqlmesh/core/model/meta.py b/sqlmesh/core/model/meta.py
index bcc87db3..b888f4d0 100644
--- a/sqlmesh/core/model/meta.py
+++ b/sqlmesh/core/model/meta.py
@@ -74,6 +74,7 @@ class ModelMeta(_Node):
     session_properties_: t.Optional[exp.Tuple] = Field(default=None, alias="session_properties")
     allow_partials: bool = False
     signals: t.List[exp.Tuple] = []
+    enabled: bool = True
 
     _bool_validator = bool_validator
     _model_kind_validator = model_kind_validator
diff --git a/sqlmesh/core/state_sync/engine_adapter.py b/sqlmesh/core/state_sync/engine_adapter.py
index 13e35337..fd3c1de6 100644
--- a/sqlmesh/core/state_sync/engine_adapter.py
+++ b/sqlmesh/core/state_sync/engine_adapter.py
@@ -940,10 +940,10 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
                 error=e,
             )
 
-            self.console.stop_migration_progress(success=False)
+            self.console.log_migration_status(success=False)
             raise SQLMeshError("SQLMesh migration failed.") from e
 
-        self.console.stop_migration_progress()
+        self.console.log_migration_status()
 
     @transactional()
     def rollback(self) -> None:
@@ -1046,7 +1046,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
 
         reversed_dag_raw = dag.reversed.graph
 
-        self.console.start_migration_progress(len(raw_snapshots))
+        self.console.start_snapshot_migration_progress(len(raw_snapshots))
 
         parsed_snapshots = LazilyParsedSnapshots(raw_snapshots)
         all_snapshot_mapping: t.Dict[SnapshotId, SnapshotTableInfo] = {}
@@ -1130,7 +1130,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             if not new_snapshot.temp_version:
                 new_snapshot.temp_version = snapshot.fingerprint.to_version()
 
-            self.console.update_migration_progress(1)
+            self.console.update_snapshot_migration_progress(1)
 
             # Visit children and evict them from the parsed_snapshots cache after.
             for child in reversed_dag_raw.get(snapshot_id, []):
@@ -1166,6 +1166,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         if new_snapshots:
             _push_new_snapshots()
 
+        self.console.stop_snapshot_migration_progress()
         return all_snapshot_mapping
 
     def _migrate_environment_rows(
@@ -1192,9 +1193,11 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
                 updated_environments.append(environment)
                 if environment.name == c.PROD:
                     updated_prod_environment = environment
+        self.console.start_env_migration_progress(len(updated_environments))
 
         for environment in updated_environments:
             self._update_environment(environment)
+            self.console.update_env_migration_progress(1)
 
         if updated_prod_environment:
             try:
@@ -1202,6 +1205,8 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             except Exception:
                 logger.warning("Failed to unpause migrated snapshots", exc_info=True)
 
+        self.console.stop_env_migration_progress()
+
     def _snapshot_ids_exist(
         self, snapshot_ids: t.Iterable[SnapshotIdLike], table_name: exp.Table
     ) -> t.Set[SnapshotId]:


Test patch:
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index ee255e6d..144397c9 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -686,7 +686,7 @@ def test_default_catalog_connections(copy_to_temp_path: t.Callable):
 def test_load_external_models(copy_to_temp_path):
     path = copy_to_temp_path("examples/sushi")
 
-    context = Context(paths=path, config="local_config")
+    context = Context(paths=path)
 
     external_model_names = [
         m.name for m in context.models.values() if m.kind.name == ModelKindName.EXTERNAL
@@ -726,3 +726,12 @@ def test_load_gateway_specific_external_models(copy_to_temp_path):
 
     # gateway explicitly set to prod; prod model should now show
     assert "prod_raw.model1" in _get_external_model_names(gateway="prod")
+
+
+def test_disabled_model(copy_to_temp_path):
+    path = copy_to_temp_path("examples/sushi")
+
+    context = Context(paths=path)
+
+    assert (path[0] / "models" / "disabled.sql").exists()
+    assert not context.get_model("sushi.disabled")
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 80724e01..15a23474 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -133,6 +133,7 @@ def test_json(snapshot: Snapshot):
             "references": [],
             "allow_partials": False,
             "signals": [],
+            "enabled": True,
         },
         "audits": [],
         "name": '"name"',
diff --git a/tests/schedulers/airflow/test_client.py b/tests/schedulers/airflow/test_client.py
index ea086a63..02506dad 100644
--- a/tests/schedulers/airflow/test_client.py
+++ b/tests/schedulers/airflow/test_client.py
@@ -118,6 +118,7 @@ def test_apply_plan(mocker: MockerFixture, snapshot: Snapshot):
                     "grains": [],
                     "allow_partials": False,
                     "signals": [],
+                    "enabled": True,
                 },
                 "audits": [],
                 "name": '"test_model"',