Add --preview flag to snow streamlit deploy for personal-workspace deploys

RELEASE-NOTES.md

@@ -21,6 +21,7 @@
 ## New additions
 * Added a `--secondary-roles` option (plus matching `SNOWFLAKE_SECONDARY_ROLES` env var and `secondary_roles` config key) to `snow connection add` and the global connection overrides. The value is forwarded to `snowflake-connector-python` and accepts `ALL` or `NONE`, so sessions can be pinned to the primary role without running an extra `USE SECONDARY ROLES` statement.
 * Added `--force` flag to `snow spcs service drop` to allow dropping services that contain block storage volumes.
+* Added `--preview` flag to `snow streamlit deploy`. Bundles local artifacts and uploads them to a per-entity subfolder under the user's default workspace live version (`snow://workspace/USER$.PUBLIC.DEFAULT$/versions/live/<entity_id>/`), then creates the Streamlit at `user$.public.<name>` with `CREATE_CODE_STAGE = FALSE`. Incompatible with `--legacy` and `--prune`.
 
 ## Fixes and improvements
 * Fixed `SELECT *` output being corrupted when joined tables share column names. Duplicate column names are now disambiguated by appending a numeric suffix (e.g. `NAME`, `NAME_2`).

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

@@ -44,7 +44,7 @@
 from snowflake.cli.api.console.console import CliConsole
 from snowflake.cli.api.constants import ObjectType
 from snowflake.cli.api.entities.utils import EntityActions
-from snowflake.cli.api.exceptions import NoProjectDefinitionError
+from snowflake.cli.api.exceptions import CliError, NoProjectDefinitionError
 from snowflake.cli.api.identifiers import FQN
 from snowflake.cli.api.output.types import (
     CommandResult,
@@ -134,6 +134,19 @@ def _check_file_exists_if_not_default(ctx: click.Context, value):
     is_flag=True,
 )
 
+PreviewOption = typer.Option(
+    False,
+    "--preview",
+    help=(
+        "Deploy the Streamlit as a personal preview in `user$.public.<name>` "
+        "sourced from your default workspace at "
+        "`snow://workspace/USER$.PUBLIC.DEFAULT$/versions/live`. "
+        "Local artifacts are uploaded to a per-entity subfolder under that "
+        "workspace live version. Incompatible with `--legacy` and `--prune`."
+    ),
+    is_flag=True,
+)
+
 
 @app.command("deploy", requires_connection=True)
 @with_project_definition()
@@ -147,6 +160,7 @@ def streamlit_deploy(
     entity_id: str = entity_argument("streamlit"),
     open_: bool = OpenOption,
     legacy: bool = LegacyOption,
+    preview: bool = PreviewOption,
     **options,
 ) -> CommandResult:
     """
@@ -156,6 +170,14 @@ def streamlit_deploy(
     in snowflake.yml and no entity_id is provided then command will raise an error.
     """
 
+    if preview and legacy:
+        raise CliError("--preview is not compatible with --legacy.")
+    if preview and prune:
+        raise CliError(
+            "--prune is not supported with --preview because the workspace "
+            "live version is shared with other files."
+        )
+
     cli_context = get_cli_context()
     workspace_ctx = _get_current_workspace_context()
 
@@ -194,6 +216,7 @@ def streamlit_deploy(
         replace=replace,
         legacy=legacy,
         prune=prune,
+        preview=preview,
     )
 
     if open_:

src/snowflake/cli/_plugins/streamlit/streamlit_entity.py

@@ -8,6 +8,9 @@
 from snowflake.cli._plugins.stage.manager import StageManager
 from snowflake.cli._plugins.streamlit.manager import StreamlitManager
 from snowflake.cli._plugins.streamlit.streamlit_entity_model import (
+    PREVIEW_DATABASE,
+    PREVIEW_SCHEMA,
+    PREVIEW_WORKSPACE_STAGE_URI,
     SPCS_RUNTIME_V2_NAME,
     StreamlitEntityModel,
 )
@@ -99,14 +102,23 @@ def deploy(
         prune: bool = False,
         bundle_map: Optional[BundleMap] = None,
         legacy: bool = False,
+        preview: bool = False,
         *args,
         **kwargs,
     ):
+        if preview and legacy:
+            raise CliError("--preview is not compatible with --legacy.")
+        if preview and prune:
+            raise CliError("--prune is not supported with --preview.")
+
         if (
             bundle_map is None
         ):  # TODO: maybe we could hold bundle map as a cached property?
             bundle_map = self.bundle()
 
+        if preview:
+            return self._deploy_preview(bundle_map=bundle_map, replace=replace)
+
         console = self._workspace_ctx.console
         console.step(f"Checking if object exists")
         object_exists = self._object_exists()
@@ -220,6 +232,50 @@ def get_deploy_sql(
 
         return query + ";"
 
+    def _preview_identifier(self) -> str:
+        """Personal-preview FQN as a bare identifier (no IDENTIFIER('...') wrap).
+
+        Wrapping in ``IDENTIFIER('...')`` interacts awkwardly with the ``$``
+        sigil in ``user$``, so we interpolate directly.
+        """
+        return f"{PREVIEW_DATABASE}.{PREVIEW_SCHEMA}.{self._entity_model.fqn.name}"
+
+    def get_preview_deploy_sql(self, replace: bool = False) -> str:
+        """Build the CREATE STREAMLIT statement for a personal-preview deploy.
+
+        ``MAIN_FILE`` is auto-prefixed with the entity_id so it points at the
+        per-entity workspace subfolder where artifacts were uploaded.
+        """
+        verb = "CREATE OR REPLACE STREAMLIT" if replace else "CREATE STREAMLIT"
+        main_file_with_prefix = f"{self.entity_id}/{self._entity_model.main_file}"
+
+        parts = [
+            f"{verb} {self._preview_identifier()}",
+            f"FROM '{PREVIEW_WORKSPACE_STAGE_URI}'",
+            f"MAIN_FILE = '{main_file_with_prefix}'",
+        ]
+
+        if self.model.imports:
+            parts.append(self.model.get_imports_sql())
+        if self.model.query_warehouse:
+            parts.append(f"QUERY_WAREHOUSE = {self.model.query_warehouse}")
+        if self.model.title:
+            parts.append(f"TITLE = '{self.model.title}'")
+        if self.model.comment:
+            parts.append(f"COMMENT = '{self.model.comment}'")
+        if self.model.external_access_integrations:
+            parts.append(self.model.get_external_access_integrations_sql())
+        if self.model.secrets:
+            parts.append(self.model.get_secrets_sql())
+
+        parts.append("CREATE_CODE_STAGE = FALSE")
+
+        if self._is_spcs_runtime_v2_mode():
+            parts.append(f"RUNTIME_NAME = '{self.model.runtime_name}'")
+            parts.append(f"COMPUTE_POOL = '{self.model.compute_pool}'")
+
+        return "\n".join(parts) + ";"
+
     def get_describe_sql(self) -> str:
         return f"DESCRIBE STREAMLIT {self._get_sql_identifier()};"
 
@@ -325,3 +381,89 @@ def _deploy_versioned(
         )
 
         StreamlitManager(connection=self._conn).grant_privileges(self.model)
+
+    def _preview_object_exists(self) -> bool:
+        """Existence check against the preview FQN (``user$.public.<name>``).
+
+        Cannot reuse ``_object_exists`` because that uses the yaml FQN.
+        """
+        sql = f"DESCRIBE STREAMLIT {self._preview_identifier()};"
+        try:
+            self._execute_query(sql, cursor_class=DictCursor)
+            return True
+        except ProgrammingError:
+            return False
+
+    def _get_preview_url(self) -> str:
+        # `user$` is a server-side alias resolved to `user$<current_user>`
+        # at SQL execution time. Snowsight URLs need the resolved name, so
+        # query CURRENT_USER() here and fall back to the literal `user$`
+        # if resolution fails (URL is informational; deploy already succeeded).
+        database = PREVIEW_DATABASE
+        try:
+            cursor = self._execute_query(
+                "SELECT 'USER$' || CURRENT_USER() AS personal_database"
+            )
+            row = cursor.fetchone()
+            # Empty CURRENT_USER() yields the literal 'USER$'; treat as unresolved.
+            if row and row[0] and not row[0].endswith("$"):
+                database = str(row[0])
+        except Exception:  # noqa: BLE001
+            log.warning(
+                "Could not resolve personal database; falling back to %r in URL.",
+                PREVIEW_DATABASE,
+                exc_info=True,
+            )
+
+        preview_fqn = FQN(
+            database=database,
+            schema=PREVIEW_SCHEMA,
+            name=self._entity_model.fqn.name,
+        ).using_connection(self._conn)
+        return make_snowsight_url(
+            self._conn, f"/#/streamlit-apps/{preview_fqn.url_identifier}"
+        )
+
+    def _deploy_preview(self, bundle_map: BundleMap, replace: bool = False) -> str:
+        console = self._workspace_ctx.console
+        preview_identifier = self._preview_identifier()
+
+        console.step(f"Checking if preview Streamlit {preview_identifier} exists")
+        if self._preview_object_exists() and not replace:
+            raise ClickException(
+                f"Preview Streamlit {preview_identifier} already exists. "
+                f"Use --replace to overwrite."
+            )
+
+        workspace_subfolder = f"{PREVIEW_WORKSPACE_STAGE_URI}/{self.entity_id}"
+        console.step(f"Uploading artifacts to {workspace_subfolder}")
+        sync_deploy_root_with_stage(
+            console=console,
+            deploy_root=bundle_map.deploy_root(),
+            bundle_map=bundle_map,
+            prune=False,  # workspace HEAD is shared; never prune unrelated files
+            recursive=True,
+            stage_path_parts=StageManager().stage_path_parts_from_str(
+                workspace_subfolder
+            ),
+            print_diff=True,
+            force_overwrite=True,
+        )
+
+        console.step(f"Creating preview Streamlit {preview_identifier}")
+        self._execute_query(self.get_preview_deploy_sql(replace=replace))
+
+        # `grants:` from snowflake.yml target the YAML FQN, not the preview
+        # FQN, so applying them here would either fail with "object does not
+        # exist" or silently grant on the wrong object. Preview deploys live
+        # in `user$.public.<name>`, which is private to the current user;
+        # users can apply grants manually if needed.
+        if self.model.grants:
+            console.warning(
+                "Skipping `grants:` in --preview mode. Preview deploys to "
+                f"{preview_identifier}, which is private to your user; apply "
+                "grants manually if needed."
+            )
+        else:
+            StreamlitManager(connection=self._conn).grant_privileges(self.model)
+        return self._get_preview_url()

src/snowflake/cli/_plugins/streamlit/streamlit_entity_model.py

@@ -28,6 +28,11 @@
 # SPCS Runtime v2 constants
 SPCS_RUNTIME_V2_NAME = "SYSTEM$ST_CONTAINER_RUNTIME_PY3_11"
 
+# Personal-preview deploy constants (used by `snow streamlit deploy --preview`)
+PREVIEW_DATABASE = "user$"
+PREVIEW_SCHEMA = "public"
+PREVIEW_WORKSPACE_STAGE_URI = "snow://workspace/USER$.PUBLIC.DEFAULT$/versions/live"
+
 
 class StreamlitEntityModel(
     EntityModelBaseWithArtifacts,

tests/streamlit/test_streamlit_commands.py

@@ -556,3 +556,74 @@ def test_execute_streamlit(self, runner, mock_streamlit_ctx):
         assert mock_streamlit_ctx.get_queries() == [
             "EXECUTE STREAMLIT IDENTIFIER('test_streamlit')()"
         ]
+
+    # ---- Personal-preview deploy (`--preview`) ----
+
+    @mock.patch(
+        "snowflake.cli._plugins.streamlit.streamlit_entity.StreamlitEntity._preview_object_exists",
+        return_value=False,
+    )
+    def test_deploy_preview_with_spcs_runtime(
+        self, _mock_preview_exists, project_directory, runner, alter_snowflake_yml
+    ):
+        """End-to-end: --preview emits the workspace-FROM SQL with SPCS runtime
+        clauses and uploads files to the per-entity workspace subfolder."""
+        with project_directory("example_streamlit_v2") as tmp_dir:
+            alter_snowflake_yml(
+                tmp_dir / "snowflake.yml",
+                parameter_path="entities.test_streamlit.runtime_name",
+                value="SYSTEM$ST_CONTAINER_RUNTIME_PY3_11",
+            )
+            alter_snowflake_yml(
+                tmp_dir / "snowflake.yml",
+                parameter_path="entities.test_streamlit.compute_pool",
+                value="MYPOOL",
+            )
+            result = runner.invoke(["streamlit", "deploy", "--preview", "--replace"])
+
+        expected_query = dedent(
+            """
+            CREATE OR REPLACE STREAMLIT user$.public.test_streamlit
+            FROM 'snow://workspace/USER$.PUBLIC.DEFAULT$/versions/live'
+            MAIN_FILE = 'test_streamlit/streamlit_app.py'
+            QUERY_WAREHOUSE = test_warehouse
+            TITLE = 'My Fancy Streamlit'
+            CREATE_CODE_STAGE = FALSE
+            RUNTIME_NAME = 'SYSTEM$ST_CONTAINER_RUNTIME_PY3_11'
+            COMPUTE_POOL = 'MYPOOL';
+            """
+        ).strip()
+
+        assert result.exit_code == 0, result.output
+        self.mock_execute.assert_any_call(expected_query)
+
+        # No ADD LIVE VERSION should be issued for preview mode.
+        executed = [c.args[0] for c in self.mock_execute.call_args_list]
+        assert not any("ADD LIVE VERSION" in q for q in executed), executed
+
+        # Files must be uploaded under the per-entity workspace subfolder.
+        upload_destinations = [
+            call.kwargs.get("stage_path")
+            or (call.args[1] if len(call.args) > 1 else None)
+            for call in self.mock_put.call_args_list
+        ]
+        assert any(
+            d
+            and "snow://workspace/USER$.PUBLIC.DEFAULT$/versions/live/test_streamlit"
+            in d
+            for d in upload_destinations
+        ), upload_destinations
+
+    def test_deploy_preview_with_prune_errors(self, project_directory, runner):
+        with project_directory("example_streamlit_v2"):
+            result = runner.invoke(["streamlit", "deploy", "--preview", "--prune"])
+
+        assert result.exit_code != 0, result.output
+        assert "--prune is not supported with --preview" in result.output
+
+    def test_deploy_preview_with_legacy_errors(self, project_directory, runner):
+        with project_directory("example_streamlit_v2"):
+            result = runner.invoke(["streamlit", "deploy", "--preview", "--legacy"])
+
+        assert result.exit_code != 0, result.output
+        assert "--preview is not compatible with --legacy" in result.output