Feat(dbt_cli): Set proper plan flags and also allow --empty and --environment

erindru · erindru · commit 1adf13dcecd0 · 2025-08-26T04:44:31.000Z
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
@@ -1296,6 +1296,7 @@ def plan(
         explain: t.Optional[bool] = None,
         ignore_cron: t.Optional[bool] = None,
         min_intervals: t.Optional[int] = None,
+        always_recreate_environment: t.Optional[bool] = None,
     ) -> Plan:
         """Interactively creates a plan.
 
@@ -1345,6 +1346,7 @@ def plan(
             explain: Whether to explain the plan instead of applying it.
             min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
                 on every model when checking for missing intervals
+            always_recreate_environment: Whether to always recreate the target environment from the `create_from` environment.
 
         Returns:
             The populated Plan object.
@@ -1376,6 +1378,7 @@ def plan(
             explain=explain,
             ignore_cron=ignore_cron,
             min_intervals=min_intervals,
+            always_recreate_environment=always_recreate_environment,
         )
 
         plan = plan_builder.build()
@@ -1428,6 +1431,7 @@ def plan_builder(
         explain: t.Optional[bool] = None,
         ignore_cron: t.Optional[bool] = None,
         min_intervals: t.Optional[int] = None,
+        always_recreate_environment: t.Optional[bool] = None,
     ) -> PlanBuilder:
         """Creates a plan builder.
 
@@ -1466,7 +1470,7 @@ def plan_builder(
             diff_rendered: Whether the diff should compare raw vs rendered models
             min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
                 on every model when checking for missing intervals
-
+            always_recreate_environment: Whether to always recreate the target environment from the `create_from` environment.
         Returns:
             The plan builder.
         """
@@ -1497,6 +1501,7 @@ def plan_builder(
             "diff_rendered": diff_rendered,
             "skip_linter": skip_linter,
             "min_intervals": min_intervals,
+            "always_recreate_environment": always_recreate_environment,
         }
         user_provided_flags: t.Dict[str, UserProvidedFlags] = {
             k: v for k, v in kwargs.items() if v is not None
@@ -1588,7 +1593,8 @@ def plan_builder(
             or (backfill_models is not None and not backfill_models),
             ensure_finalized_snapshots=self.config.plan.use_finalized_state,
             diff_rendered=diff_rendered,
-            always_recreate_environment=self.config.plan.always_recreate_environment,
+            always_recreate_environment=always_recreate_environment
+            or self.config.plan.always_recreate_environment,
         )
         modified_model_names = {
             *context_diff.modified_snapshots,
diff --git a/sqlmesh_dbt/cli.py b/sqlmesh_dbt/cli.py
@@ -81,7 +81,15 @@ def dbt(
     "--full-refresh",
     help="If specified, dbt will drop incremental models and fully-recalculate the incremental table from the model definition.",
 )
+@click.option(
+    "--environment",
+    help="Run against a specific Virtual Data Environment (VDE) instead of the main environment",
+)
+@click.option(
+    "--empty/--no-empty", default=False, help="If specified, limit input refs and sources"
+)
 @vars_option
+@cli_global_error_handler
 @click.pass_context
 def run(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.Any) -> None:
     """Compile SQL and execute against the current target database."""
@@ -92,6 +100,7 @@ def run(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.An
 @select_option
 @exclude_option
 @vars_option
+@cli_global_error_handler
 @click.pass_context
 def list_(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.Any) -> None:
     """List the resources in your project"""
diff --git a/sqlmesh_dbt/operations.py b/sqlmesh_dbt/operations.py
@@ -11,6 +11,7 @@
     from sqlmesh.dbt.project import Project
     from sqlmesh_dbt.console import DbtCliConsole
     from sqlmesh.core.model import Model
+    from sqlmesh.core.plan import Plan
 
 logger = logging.getLogger(__name__)
 
@@ -34,21 +35,20 @@ def list_(
 
     def run(
         self,
+        environment: t.Optional[str] = None,
         select: t.Optional[t.List[str]] = None,
         exclude: t.Optional[t.List[str]] = None,
         full_refresh: bool = False,
-    ) -> None:
-        select_models = None
-
-        if sqlmesh_selector := selectors.to_sqlmesh(select or [], exclude or []):
-            select_models = [sqlmesh_selector]
-
-        self.context.plan(
-            select_models=select_models,
-            run=True,
-            no_diff=True,
-            no_prompts=True,
-            auto_apply=True,
+        empty: bool = False,
+    ) -> Plan:
+        return self.context.plan(
+            **self._plan_options(
+                environment=environment,
+                select=select,
+                exclude=exclude,
+                full_refresh=full_refresh,
+                empty=empty,
+            )
         )
 
     def _selected_models(
@@ -66,6 +66,88 @@ def _selected_models(
 
         return selected_models
 
+    def _plan_options(
+        self,
+        environment: t.Optional[str] = None,
+        select: t.Optional[t.List[str]] = None,
+        exclude: t.Optional[t.List[str]] = None,
+        empty: bool = False,
+        full_refresh: bool = False,
+    ) -> t.Dict[str, t.Any]:
+        import sqlmesh.core.constants as c
+
+        # convert --select and --exclude to a selector expression for the SQLMesh selector engine
+        select_models = None
+        if sqlmesh_selector := selectors.to_sqlmesh(select or [], exclude or []):
+            select_models = [sqlmesh_selector]
+
+        is_dev = environment and environment != c.PROD
+        is_prod = not is_dev
+
+        options: t.Dict[str, t.Any] = {}
+
+        if is_prod or (is_dev and select_models):
+            # prod plans should "catch up" before applying the changes so that after the command finishes prod is the latest it can be
+            # dev plans *with* selectors should do the same as the user is saying "specifically update these models to the latest"
+            # dev plans *without* selectors should just have the defaults of never exceeding prod as the user is saying "just create this env" without focusing on any specific models
+            options.update(
+                dict(
+                    # always catch the data up to latest rather than only operating on what has been loaded before
+                    run=True,
+                    # don't taking cron schedules into account when deciding what models to run, do everything even if it just ran
+                    ignore_cron=True,
+                )
+            )
+
+        if is_dev:
+            options.update(
+                dict(
+                    # don't create views for all of prod in the dev environment
+                    include_unmodified=False,
+                    # always plan from scratch against prod rather than planning against the previous state of an existing dev environment
+                    # this results in the full scope of changes vs prod always being shown on the local branch
+                    create_from=c.PROD,
+                    always_recreate_environment=True,
+                    # setting enable_preview=None enables dev previews of forward_only changes for dbt projects IF the target engine supports cloning
+                    # if we set enable_preview=True here, this enables dev previews in all cases.
+                    # In the case of dbt default INCREMENTAL_UNMANAGED models, this will cause incremental models to be fully rebuilt (potentially a very large computation)
+                    # just to have the results thrown away on promotion to prod because dev previews are not promotable.
+                    #
+                    # TODO: if the user "upgrades" to an INCREMENTAL_BY_TIME_RANGE by defining a "time_column", we can inject leak guards to compute
+                    # just a preview instead of the whole thing like we would in a native project, but the enable_preview setting is at the plan level
+                    # and not the individual model level so we currently have no way of doing this selectively
+                    enable_preview=None,
+                )
+            )
+
+        if empty:
+            # dbt --empty adds LIMIT 0 to the queries, resulting in empty tables
+            # this lines up with --skip-backfill in SQLMesh
+            options["skip_backfill"] = True
+
+            if is_prod:
+                # to prevent the following error:
+                # > ConfigError: When targeting the production environment either the backfill should not be skipped or
+                # > the lack of data gaps should be enforced (--no-gaps flag).
+                options["no_gaps"] = True
+
+        if full_refresh:
+            # TODO: handling this requires some updates in the engine to enable restatements+changes in the same plan without affecting prod
+            # if the plan targets dev
+            pass
+
+        return dict(
+            environment=environment,
+            select_models=select_models,
+            # dont output a diff of model changes
+            no_diff=True,
+            # don't throw up any prompts like "set the effective date" - use defaults
+            no_prompts=True,
+            # start doing work immediately (since no_diff is set, there isnt really anything for the user to say yes/no to)
+            auto_apply=True,
+            **options,
+        )
+
     @property
     def console(self) -> DbtCliConsole:
         console = self.context.console
diff --git a/tests/dbt/cli/test_operations.py b/tests/dbt/cli/test_operations.py
@@ -1,13 +1,35 @@
+import typing as t
 from pathlib import Path
 import pytest
 from sqlmesh_dbt.operations import create
 from sqlmesh.utils import yaml
 from sqlmesh.utils.errors import SQLMeshError
 import time_machine
+from sqlmesh.core.console import NoopConsole
+from sqlmesh.core.plan import PlanBuilder
 
 pytestmark = pytest.mark.slow
 
 
+class PlanCapturingConsole(NoopConsole):
+    def plan(
+        self,
+        plan_builder: PlanBuilder,
+        auto_apply: bool,
+        default_catalog: t.Optional[str],
+        no_diff: bool = False,
+        no_prompts: bool = False,
+    ) -> None:
+        self.plan_builder = plan_builder
+        self.auto_apply = auto_apply
+        self.default_catalog = default_catalog
+        self.no_diff = no_diff
+        self.no_prompts = no_prompts
+
+        # normal console starts applying the plan here; we dont because we just want to capture the parameters
+        # and check they were set correctly
+
+
 def test_create_sets_and_persists_default_start_date(jaffle_shop_duckdb: Path):
     with time_machine.travel("2020-01-02 00:00:00 UTC"):
         from sqlmesh.utils.date import yesterday_ds, to_ds
@@ -83,3 +105,136 @@ def test_create_can_set_project_variables(jaffle_shop_duckdb: Path):
     query = test_model.render_query()
     assert query is not None
     assert query.sql() == "SELECT 'bar' AS \"a\""
+
+
+def test_run_option_mapping(jaffle_shop_duckdb: Path):
+    operations = create(project_dir=jaffle_shop_duckdb)
+    console = PlanCapturingConsole()
+    operations.context.console = console
+
+    plan = operations.run()
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill is None
+    assert {s.name for s in plan.snapshots} == {k for k in operations.context.snapshots}
+
+    plan = operations.run(select=["main.stg_orders+"])
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill == {
+        '"jaffle_shop"."main"."customers"',
+        '"jaffle_shop"."main"."orders"',
+        '"jaffle_shop"."main"."stg_orders"',
+    }
+    assert {s.name for s in plan.snapshots} == plan.selected_models_to_backfill
+
+    plan = operations.run(select=["main.stg_orders+"], exclude=["main.customers"])
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill == {
+        '"jaffle_shop"."main"."orders"',
+        '"jaffle_shop"."main"."stg_orders"',
+    }
+    assert {s.name for s in plan.snapshots} == plan.selected_models_to_backfill
+
+    plan = operations.run(exclude=["main.customers"])
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill == {k for k in operations.context.snapshots} - {
+        '"jaffle_shop"."main"."customers"'
+    }
+    assert {s.name for s in plan.snapshots} == plan.selected_models_to_backfill
+
+    plan = operations.run(empty=True)
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is True
+    assert plan.selected_models_to_backfill is None
+    assert {s.name for s in plan.snapshots} == {k for k in operations.context.snapshots}
+
+
+def test_run_option_mapping_dev(jaffle_shop_duckdb: Path):
+    # create prod so that dev has something to compare against
+    operations = create(project_dir=jaffle_shop_duckdb)
+    operations.run()
+
+    (jaffle_shop_duckdb / "models" / "new_model.sql").write_text("select 1")
+
+    operations = create(project_dir=jaffle_shop_duckdb)
+
+    console = PlanCapturingConsole()
+    operations.context.console = console
+
+    plan = operations.run(environment="dev")
+    assert plan.environment.name == "dev"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.include_unmodified is False
+    assert plan.context_diff.create_from == "prod"
+    assert plan.context_diff.is_new_environment is True
+    assert (
+        console.plan_builder._enable_preview is False
+    )  # duckdb doesnt support cloning so dev previews are not enabled for dbt projects
+    assert plan.end_bounded is True
+    assert plan.ignore_cron is False
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill == {'"jaffle_shop"."main"."new_model"'}
+
+    plan = operations.run(environment="dev", empty=True)
+    assert plan.environment.name == "dev"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.include_unmodified is False
+    assert plan.context_diff.create_from == "prod"
+    assert plan.context_diff.is_new_environment is True
+    assert console.plan_builder._enable_preview is False
+    assert plan.end_bounded is True
+    assert plan.ignore_cron is False
+    assert plan.skip_backfill is True
+    assert plan.selected_models_to_backfill == {'"jaffle_shop"."main"."new_model"'}
+
+    plan = operations.run(environment="dev", select=["main.stg_orders+"])
+    assert plan.environment.name == "dev"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.include_unmodified is False
+    assert plan.context_diff.create_from == "prod"
+    assert plan.context_diff.is_new_environment is True
+    assert console.plan_builder._enable_preview is False
+    # dev plans with --select have run=True, ignore_cron=True set
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    # note: the new model in the dev environment is ignored in favour of the explicitly selected ones
+    assert plan.selected_models_to_backfill == {
+        '"jaffle_shop"."main"."customers"',
+        '"jaffle_shop"."main"."orders"',
+        '"jaffle_shop"."main"."stg_orders"',
+    }
