simplify

Author: izeigerman

sqlmesh/core/plan/stages.py

@@ -281,9 +281,7 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
             snapshots_with_schema_migration = [
                 snapshots[s_id]
                 for s_id in dag.subdag(*snapshot_ids_with_schema_migration)
-                if snapshots[s_id].is_paused
-                and snapshots[s_id].is_model
-                and not snapshots[s_id].is_symbolic
+                if snapshots[s_id].supports_schema_migration_in_prod
             ]
 
         snapshots_to_intervals = self._missing_intervals(

sqlmesh/core/snapshot/definition.py

@@ -1477,19 +1477,19 @@ def expiration_ts(self) -> int:
             check_categorical_relative_expression=False,
         )
 
+    @property
+    def supports_schema_migration_in_prod(self) -> bool:
+        """Returns whether or not this snapshot supports schema migration when deployed to production."""
+        return self.is_paused and self.is_model and not self.is_symbolic
+
     @property
     def requires_schema_migration_in_prod(self) -> bool:
         """Returns whether or not this snapshot requires a schema migration when deployed to production."""
-        return (
-            self.is_paused
-            and self.is_model
-            and not self.is_symbolic
-            and (
-                (self.previous_version and self.previous_version.version == self.version)
-                or self.model.forward_only
-                or bool(self.model.physical_version)
-                or not self.virtual_environment_mode.is_full
-            )
+        return self.supports_schema_migration_in_prod and (
+            (self.previous_version and self.previous_version.version == self.version)
+            or self.model.forward_only
+            or bool(self.model.physical_version)
+            or not self.virtual_environment_mode.is_full
         )
 
     @property