feat: [SNOW-3444080] add env.yml support to snow dbt deploy

Adds three flags to `snow dbt deploy` that bring env.yml into the deploy
flow alongside profiles.yml:

- `--env-file-dir <path>`: inject env.yml from a separate directory into
  the staged project root, mirroring `--profiles-dir`.
- `--default-environment <name>`: set DEFAULT_ENVIRONMENT on the DBT
  PROJECT object (CREATE / CREATE OR REPLACE / ALTER SET).
- `--unset-default-environment`: clear DEFAULT_ENVIRONMENT on alter,
  mutually exclusive with --default-environment.

env.yml is staged like profiles.yml: copied from the source directory by
default, optionally overridden via `--env-file-dir`, and round-tripped
through pyyaml so YAML comments don't leak onto the stage. The CLI does
no client-side schema validation — the server enforces env-var prefixes,
secret-key rules, and reserved environment names.

Server support for these properties is gated by ENABLE_DBT_ENV_VARS
(GS PR #404443).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: sfc-gh-yuding

src/snowflake/cli/_plugins/dbt/commands.py

@@ -21,6 +21,7 @@
 from click import types
 from snowflake.cli._plugins.dbt.constants import (
     DBT_COMMANDS,
+    ENV_FILENAME,
     OUTPUT_COLUMN_NAME,
     PROFILES_FILENAME,
     RESULT_COLUMN_NAME,
@@ -72,6 +73,16 @@
     "--unset-default-target",
     mutually_exclusive=["default_target"],
 )
+DefaultEnvironmentOption = OverrideableOption(
+    None,
+    "--default-environment",
+    mutually_exclusive=["unset_default_environment"],
+)
+UnsetDefaultEnvironmentOption = OverrideableOption(
+    False,
+    "--unset-default-environment",
+    mutually_exclusive=["default_environment"],
+)
 
 add_object_command_aliases(
     app=app,
@@ -89,6 +100,11 @@ def _env_callback(value: Optional[str]) -> Optional[str]:
     return _reject_control_chars(value, "--env")
 
 
+def _default_environment_callback(value: Optional[str]) -> Optional[str]:
+    return _reject_control_chars(value, "--default-environment")
+
+
+
 @app.command(
     "deploy",
     requires_connection=True,
@@ -105,6 +121,16 @@ def deploy_dbt(
         show_default=False,
         default=None,
     ),
+    env_file_dir: Optional[str] = typer.Option(
+        help=(
+            f"Path to directory containing {ENV_FILENAME}. If provided, the file is "
+            f"injected into the deployed project root, overwriting any {ENV_FILENAME} "
+            f"present in --source."
+        ),
+        show_default=False,
+        default=None,
+        hidden=not FeatureFlag.ENABLE_DBT_PROJECT_ENV_VARS.is_enabled(),
+    ),
     force: Optional[bool] = typer.Option(
         False,
         help="Overwrites conflicting files in the project, if any.",
@@ -115,6 +141,20 @@ def deploy_dbt(
     unset_default_target: Optional[bool] = UnsetDefaultTargetOption(
         help="Unset the default target for the dbt project. Mutually exclusive with --default-target.",
     ),
+    default_environment: Optional[str] = DefaultEnvironmentOption(
+        help=(
+            f"Default environment for the dbt project (DEFAULT_ENVIRONMENT). "
+            f"Selects the environment block from {ENV_FILENAME} that the project "
+            f"compiles and executes with by default. "
+            f"Mutually exclusive with --unset-default-environment."
+        ),
+        callback=_default_environment_callback,
+        hidden=not FeatureFlag.ENABLE_DBT_PROJECT_ENV_VARS.is_enabled(),
+    ),
+    unset_default_environment: Optional[bool] = UnsetDefaultEnvironmentOption(
+        help="Unset the default environment for the dbt project. Mutually exclusive with --default-environment.",
+        hidden=not FeatureFlag.ENABLE_DBT_PROJECT_ENV_VARS.is_enabled(),
+    ),
     external_access_integrations: Optional[list[str]] = typer.Option(
         None,
         "--external-access-integration",
@@ -144,9 +184,12 @@ def deploy_dbt(
     """
     project_path = SecurePath(source) if source is not None else SecurePath.cwd()
     profiles_dir_path = SecurePath(profiles_dir) if profiles_dir else project_path
+    env_file_path = SecurePath(env_file_dir) if env_file_dir else None
     attrs = DBTDeployAttributes(
         default_target=default_target,
         unset_default_target=unset_default_target,
+        default_environment=default_environment,
+        unset_default_environment=unset_default_environment,
         external_access_integrations=external_access_integrations,
         install_local_deps=install_local_deps,
         dbt_version=dbt_version,
@@ -156,6 +199,7 @@ def deploy_dbt(
             name,
             path=project_path.resolve(),
             profiles_path=profiles_dir_path.resolve(),
+            env_file_path=env_file_path.resolve() if env_file_path else None,
             force=force,
             attrs=attrs,
         )

src/snowflake/cli/_plugins/dbt/constants.py

@@ -16,6 +16,7 @@
 OUTPUT_COLUMN_NAME = "STDOUT"
 PROFILES_FILENAME = "profiles.yml"
 SUPPORTED_DBT_VERSIONS_QUERY = "SELECT SYSTEM$SUPPORTED_DBT_VERSIONS()"
+ENV_FILENAME = "env.yml"
 
 DBT_COMMANDS = [
     "build",

src/snowflake/cli/_plugins/dbt/manager.py

@@ -24,6 +24,7 @@
 
 import yaml
 from snowflake.cli._plugins.dbt.constants import (
+    ENV_FILENAME,
     PROFILES_FILENAME,
     SUPPORTED_DBT_VERSIONS_QUERY,
 )
@@ -86,6 +87,7 @@ def _no_duplicates_constructor(loader, node, deep=False):
 
 class DBTObjectEditableAttributes(TypedDict):
     default_target: Optional[str]
+    default_environment: Optional[str]
     external_access_integrations: Optional[List[str]]
     dbt_version: Optional[str]
 
@@ -96,6 +98,8 @@ class DBTDeployAttributes:
 
     default_target: Optional[str] = None
     unset_default_target: bool = False
+    default_environment: Optional[str] = None
+    unset_default_environment: bool = False
     external_access_integrations: Optional[List[str]] = None
     install_local_deps: bool = False
     dbt_version: Optional[str] = None
@@ -152,6 +156,7 @@ def get_dbt_object_attributes(name: FQN) -> Optional[DBTObjectEditableAttributes
 
         return DBTObjectEditableAttributes(
             default_target=row_dict.get("default_target"),
+            default_environment=row_dict.get("default_environment"),
             external_access_integrations=external_access_integrations,
             dbt_version=row_dict.get("dbt_version"),
         )
@@ -197,6 +202,7 @@ def deploy(
         profiles_path: SecurePath,
         force: bool,
         attrs: DBTDeployAttributes,
+        env_file_path: Optional[SecurePath] = None,
     ) -> SnowflakeCursor:
         dbt_project_path = path / "dbt_project.yml"
         if not dbt_project_path.exists():
@@ -216,6 +222,13 @@ def deploy(
         if attrs.dbt_version:
             self._validate_dbt_version(attrs.dbt_version)
 
+        if env_file_path is not None:
+            env_file = env_file_path / ENV_FILENAME
+            if not env_file.exists():
+                raise CliError(
+                    f"{ENV_FILENAME} does not exist in directory {env_file_path.path.absolute()}."
+                )
+
         with cli_console.phase("Creating temporary stage"):
             stage_manager = StageManager()
             stage_fqn = FQN.from_resource(ObjectType.DBT_PROJECT, fqn, "STAGE")
@@ -227,6 +240,8 @@ def deploy(
                 tmp_path = Path(tmp)
                 stage_manager.copy_to_tmp_dir(path.path, tmp_path)
                 self._prepare_profiles_file(profiles_path.path, tmp_path)
+                if env_file_path is not None:
+                    self._prepare_env_file(env_file_path.path, tmp_path)
                 result_count = len(
                     list(
                         stage_manager.put_recursive(
@@ -267,6 +282,15 @@ def _deploy_alter(
         ):
             set_properties.append(f"DEFAULT_TARGET='{attrs.default_target}'")
 
+        # Always issue SET/UNSET when the user asks; the server treats
+        # UNSET-on-null as a no-op.
+        if attrs.unset_default_environment:
+            unset_properties.append("DEFAULT_ENVIRONMENT")
+        elif attrs.default_environment:
+            set_properties.append(
+                f"DEFAULT_ENVIRONMENT={to_string_literal(attrs.default_environment)}"
+            )
+
         # Comparing dbt_version to existing project's dbt_version might be ambiguous
         # if previously project was locked to just minor version and now user wants to
         # lock it to a patch as well. If target version is provided, it's better to just
@@ -335,6 +359,10 @@ def _deploy_create(
         query += f"\nFROM {stage_name}"
         if attrs.default_target:
             query += f" DEFAULT_TARGET='{attrs.default_target}'"
+        if attrs.default_environment:
+            query += (
+                f" DEFAULT_ENVIRONMENT={to_string_literal(attrs.default_environment)}"
+            )
         if attrs.dbt_version:
             query += f" DBT_VERSION={to_string_literal(attrs.dbt_version)}"
         query = self._handle_external_access_integrations_query(
@@ -366,6 +394,10 @@ def _deploy_create_or_replace(
         query += f"\nFROM {stage_name}"
         if attrs.default_target:
             query += f" DEFAULT_TARGET='{attrs.default_target}'"
+        if attrs.default_environment:
+            query += (
+                f" DEFAULT_ENVIRONMENT={to_string_literal(attrs.default_environment)}"
+            )
         if attrs.dbt_version:
             query += f" DBT_VERSION={to_string_literal(attrs.dbt_version)}"
         query = self._handle_external_access_integrations_query(
@@ -462,6 +494,17 @@ def _prepare_profiles_file(profiles_path: Path, tmp_path: Path):
         ) as sfd, target_profiles_file.open(mode="w") as tfd:
             yaml.safe_dump(yaml.safe_load(sfd), tfd)
 
+    @staticmethod
+    def _prepare_env_file(env_path: Path, tmp_path: Path):
+        source_env_file = SecurePath(env_path / ENV_FILENAME)
+        target_env_file = SecurePath(tmp_path / ENV_FILENAME)
+        if target_env_file.exists():
+            target_env_file.unlink()
+        with source_env_file.open(
+            read_file_limit_mb=DEFAULT_SIZE_LIMIT_MB
+        ) as sfd, target_env_file.open(mode="w") as tfd:
+            yaml.safe_dump(yaml.safe_load(sfd), tfd)
+
     def execute(
         self,
         dbt_command: str,

tests/dbt/conftest.py

@@ -2,7 +2,7 @@
 
 import pytest
 import yaml
-from snowflake.cli._plugins.dbt.constants import PROFILES_FILENAME
+from snowflake.cli._plugins.dbt.constants import ENV_FILENAME, PROFILES_FILENAME
 
 
 @pytest.fixture
@@ -40,6 +40,19 @@ def profile():
     return profiles
 
 
+@pytest.fixture()
+def env_yml():
+    return {
+        "env_config": {
+            "default_environment": "dev",
+            "environments": [
+                {"name": "dev", "env": {"DBT_FOO": "bar"}},
+                {"name": "prod", "env": {"DBT_FOO": "baz"}},
+            ],
+        }
+    }
+
+
 @pytest.fixture
 def project_path(tmp_path_factory):
     source_path = tmp_path_factory.mktemp("dbt_project")
@@ -55,6 +68,13 @@ def dbt_project_path(project_path, profile):
     yield project_path
 
 
+@pytest.fixture
+def env_yml_dir(tmp_path_factory, env_yml):
+    env_path = tmp_path_factory.mktemp("dbt_envs")
+    (env_path / ENV_FILENAME).write_text(yaml.dump(env_yml))
+    yield env_path
+
+
 @pytest.fixture
 def mock_get_dbt_object_attributes():
     with mock.patch(

tests/dbt/test_dbt_commands.py

@@ -194,6 +194,102 @@ def test_dbt_deploy_with_custom_profiles_dir(
         assert str(mock_deploy.call_args[0][0]) == "TEST_PIPELINE"
         assert call_kwargs["profiles_path"] == SecurePath(new_profiles_directory)
 
+    def test_dbt_deploy_with_env_file_dir(
+        self, runner, dbt_project_path, env_yml_dir, mock_deploy
+    ):
+        result = runner.invoke(
+            [
+                "dbt",
+                "deploy",
+                "TEST_PIPELINE",
+                f"--source={dbt_project_path}",
+                f"--env-file-dir={env_yml_dir}",
+            ]
+        )
+
+        assert result.exit_code == 0, result.output
+        mock_deploy.assert_called_once()
+        call_kwargs = mock_deploy.call_args[1]
+        assert call_kwargs["env_file_path"] == SecurePath(env_yml_dir)
+
+    def test_dbt_deploy_without_env_file_dir_passes_none(
+        self, runner, dbt_project_path, mock_deploy
+    ):
+        result = runner.invoke(
+            [
+                "dbt",
+                "deploy",
+                "TEST_PIPELINE",
+                f"--source={dbt_project_path}",
+            ]
+        )
+
+        assert result.exit_code == 0, result.output
+        mock_deploy.assert_called_once()
+        call_kwargs = mock_deploy.call_args[1]
+        assert call_kwargs["env_file_path"] is None
+
+    def test_deploy_with_default_environment_passes_to_manager(
+        self, runner, dbt_project_path, mock_deploy
+    ):
+        result = runner.invoke(
+            [
+                "dbt",
+                "deploy",
+                "TEST_PIPELINE",
+                f"--source={dbt_project_path}",
+                "--default-environment=dev",
+            ]
+        )
+
+        assert result.exit_code == 0, result.output
+        mock_deploy.assert_called_once()
+        call_kwargs = mock_deploy.call_args[1]
+        assert call_kwargs["attrs"].default_environment == "dev"
+        assert call_kwargs["attrs"].unset_default_environment is False
+
+    def test_deploy_with_unset_default_environment_passes_to_manager(
+        self, runner, dbt_project_path, mock_deploy
+    ):
+        result = runner.invoke(
+            [
+                "dbt",
+                "deploy",
+                "TEST_PIPELINE",
+                f"--source={dbt_project_path}",
+                "--unset-default-environment",
+            ]
+        )
+
+        assert result.exit_code == 0, result.output
+        mock_deploy.assert_called_once()
+        call_kwargs = mock_deploy.call_args[1]
+        assert call_kwargs["attrs"].default_environment is None
+        assert call_kwargs["attrs"].unset_default_environment is True
+
+    def test_deploy_with_both_default_environment_and_unset_default_environment_fails(
+        self,
+        mock_connect,
+        runner,
+        dbt_project_path,
+    ):
+        result = runner.invoke(
+            [
+                "dbt",
+                "deploy",
+                "TEST_PIPELINE",
+                f"--source={dbt_project_path}",
+                "--default-environment=dev",
+                "--unset-default-environment",
+            ]
+        )
+
+        assert result.exit_code == 2, result.output
+        # Box-rendered error wraps across lines; check the key parts.
+        assert "'--unset-default-environment'" in result.output
+        assert "'--default-environment'" in result.output
+        assert "incompatible" in result.output
+
     def test_deploy_with_default_target_passes_to_manager(
         self, runner, dbt_project_path, mock_deploy
     ):