Fixes

src/snowflake/cli/_plugins/nativeapp/entities/application_package.py

@@ -833,6 +833,7 @@ def _bundle_children(self, action_ctx: ActionContext) -> List[str]:
                 child_entity.get_deploy_sql(
                     artifacts_dir=child_artifacts_dir.relative_to(self.deploy_root),
                     schema=child_schema,
+                    replace=True,
                 )
             )
             if app_role:

src/snowflake/cli/_plugins/nativeapp/feature_flags.py

@@ -18,7 +18,9 @@
 
 
 @unique
-class FeatureFlag(FeatureFlagMixin):
+class FeatureFlag(
+    FeatureFlagMixin
+):  # TODO move this to snowflake.cli.api.feature_flags
     ENABLE_NATIVE_APP_PYTHON_SETUP = BooleanFlag(
         "ENABLE_NATIVE_APP_PYTHON_SETUP", False
     )

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

@@ -1,7 +1,10 @@
 import functools
+from pathlib import Path
 from typing import Optional
 
+from click import ClickException
 from snowflake.cli._plugins.connection.util import make_snowsight_url
+from snowflake.cli._plugins.nativeapp.feature_flags import FeatureFlag
 from snowflake.cli._plugins.streamlit.streamlit_entity_model import (
     StreamlitEntityModel,
 )
@@ -16,6 +19,11 @@ class StreamlitEntity(EntityBase[StreamlitEntityModel]):
     A Streamlit app.
     """
 
+    def __init__(self, *args, **kwargs):
+        if not FeatureFlag.ENABLE_NATIVE_APP_CHILDREN.is_enabled():
+            raise NotImplementedError("Streamlit entity is not implemented yet")
+        super().__init__(*args, **kwargs)
+
     @property
     def root(self):
         return self._workspace_ctx.project_root
@@ -36,43 +44,23 @@ def _conn(self):
     def model(self):
         return self._entity_model  # noqa
 
-    def action_bundle(self, ctx: ActionContext, *args, **kwargs):
-        # get all files from the model
-        artifacts = self._entity_model.artifacts
-        # get root
-        output_folder = self.root / "output" / self._entity_model.stage
-        output_folder.mkdir(parents=True, exist_ok=True)
-
-        output_files = []
-
-        # This is far from , but will be replaced by bundlemap mappings.
-        for file in artifacts:
-            output_file = output_folder / file.name
-
-            if file.is_file():
-                SecurePath(file).copy(output_file)
-            elif file.is_dir():
-                output_file.mkdir(parents=True, exist_ok=True)
-                SecurePath(file).copy(output_file, dirs_exist_ok=True)
-
-                output_files.append(output_file)
-
-        return output_files
+    def action_bundle(self, action_ctx: ActionContext, *args, **kwargs):
+        return self.bundle()
 
     def action_deploy(self, action_ctx: ActionContext, *args, **kwargs):
         # After adding bundle map- we should use it's mapping here
 
-        query = self.get_deploy_sql(action_ctx, *args, **kwargs)
+        query = self.get_deploy_sql()
         result = self._sql_executor.execute_query(query)
         return result
 
     def action_drop(self, action_ctx: ActionContext, *args, **kwargs):
-        return self._sql_executor.execute_query(self.get_drop_sql(action_ctx))
+        return self._sql_executor.execute_query(self.get_drop_sql())
 
     def action_execute(
         self, action_ctx: ActionContext, *args, **kwargs
     ) -> SnowflakeCursor:
-        return self._sql_executor.execute_query(self.get_execute_sql(action_ctx))
+        return self._sql_executor.execute_query(self.get_execute_sql())
 
     def action_get_url(
         self, action_ctx: ActionContext, *args, **kwargs
@@ -82,15 +70,48 @@ def action_get_url(
             self._conn, f"/#/streamlit-apps/{name.url_identifier}"
         )
 
+    def bundle(self, output_dir: Optional[Path] = None):
+
+        if not output_dir:
+            output_dir = self.root / "output" / self._entity_model.stage
+
+        artifacts = self._entity_model.artifacts
+
+        output_dir.mkdir(parents=True, exist_ok=True)  # type: ignore
+
+        output_files = []
+
+        # This is far from , but will be replaced by bundlemap mappings.
+        for file in artifacts:
+            output_file = output_dir / file.name
+
+            if file.is_file():
+                SecurePath(file).copy(output_file)
+            elif file.is_dir():
+                output_file.mkdir(parents=True, exist_ok=True)
+                SecurePath(file).copy(output_file, dirs_exist_ok=True)
+
+                output_files.append(output_file)
+
+        return output_files
+
+    def action_share(
+        self, action_ctx: ActionContext, to_role: str, *args, **kwargs
+    ) -> SnowflakeCursor:
+        return self._sql_executor.execute_query(self.get_share_sql(to_role))
+
     def get_deploy_sql(
         self,
-        action_ctx: ActionContext,
         if_not_exists: bool = False,
         replace: bool = False,
         from_stage_name: Optional[str] = None,
+        artifacts_dir: Optional[Path] = None,
+        schema: Optional[str] = None,
         *args,
         **kwargs,
     ):
+        if replace and if_not_exists:
+            raise ClickException("Cannot specify both replace and if_not_exists")
 
         if replace:
             query = "CREATE OR REPLACE "
@@ -99,47 +120,48 @@ def get_deploy_sql(
         else:
             query = "CREATE "
 
-        query += f"STREAMLIT {self._entity_model.fqn.sql_identifier} \n"
+        schema_to_use = schema or self._entity_model.fqn.schema
+        query += f"STREAMLIT {self._entity_model.fqn.set_schema(schema_to_use).sql_identifier}"
 
         if from_stage_name:
-            query += f" ROOT_LOCATION = '{from_stage_name}' \n"
+            query += f"\nROOT_LOCATION = '{from_stage_name}'"
+        elif artifacts_dir:
+            query += f"\nFROM = '{artifacts_dir}'"
 
-        query += f" MAIN_FILE = '{self._entity_model.main_file}' \n"
+        query += f"\nMAIN_FILE = '{self._entity_model.main_file}'"
 
         if self.model.imports:
-            query += self.model.get_imports_sql() + "\n"
+            query += "\n" + self.model.get_imports_sql()
 
         if self.model.query_warehouse:
-            query += f" QUERY_WAREHOUSE = '{self.model.query_warehouse}' \n"
+            query += f"\nQUERY_WAREHOUSE = '{self.model.query_warehouse}'"
 
         if self.model.title:
-            query += f" TITLE = '{self.model.title}' \n"
+            query += f"\nTITLE = '{self.model.title}'"
 
         if self.model.comment:
-            query += f" COMMENT = '{self.model.comment}' \n"
+            query += f"\nCOMMENT = '{self.model.comment}'"
 
         if self.model.external_access_integrations:
-            query += self.model.get_external_access_integrations_sql() + "\n"
+            query += "\n" + self.model.get_external_access_integrations_sql()
 
         if self.model.secrets:
-            query += self.model.get_secrets_sql() + "\n"
+            query += "\n" + self.model.get_secrets_sql()
 
-        return query
+        return query + ";"
 
-    def action_share(
-        self, action_ctx: ActionContext, to_role: str, *args, **kwargs
-    ) -> SnowflakeCursor:
-        return self._sql_executor.execute_query(
-            self.get_usage_grant_sql(action_ctx, to_role)
-        )
+    def get_drop_sql(self):
+        return f"DROP STREAMLIT {self._entity_model.fqn};"
 
-    def get_drop_sql(self, action_ctx: ActionContext, *args, **kwargs):
-        return f"DROP STREAMLIT {self._entity_model.fqn}"
+    def get_execute_sql(self):
+        return f"EXECUTE STREAMLIT {self._entity_model.fqn}();"
 
-    def get_execute_sql(self, action_ctx: ActionContext, *args, **kwargs):
-        return f"EXECUTE STREAMLIT {self._entity_model.fqn}()"
+    def get_share_sql(self, to_role: str) -> str:
+        return f"grant usage on streamlit {self.model.fqn.sql_identifier} to role {to_role};"
 
-    def get_usage_grant_sql(
-        self, action_ctx: ActionContext, to_role: str, *args, **kwargs
-    ) -> str:
-        return f"GRANT USAGE ON STREAMLIT {{self._entity_model.fqn}} to role {to_role}"
+    def get_usage_grant_sql(self, app_role: str, schema: Optional[str] = None) -> str:
+        entity_id = self.entity_id
+        streamlit_name = f"{schema}.{entity_id}" if schema else entity_id
+        return (
+            f"GRANT USAGE ON STREAMLIT {streamlit_name} TO APPLICATION ROLE {app_role};"
+        )

tests/nativeapp/test_children.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from pathlib import Path
-from textwrap import dedent
 
 import pytest
 import yaml
@@ -140,13 +139,13 @@ def test_children_bundle_with_custom_dir(project_directory):
             setup_script_content = f.read()
             custom_dir_path = Path("_entities", "my_streamlit")
             assert setup_script_content.endswith(
-                dedent(
-                    f"""
-                    -- AUTO GENERATED CHILDREN SECTION
-                    CREATE OR REPLACE STREAMLIT v_schema.my_streamlit FROM '{custom_dir_path}' MAIN_FILE='streamlit_app.py';
-                    CREATE APPLICATION ROLE IF NOT EXISTS my_app_role;
-                    GRANT USAGE ON SCHEMA v_schema TO APPLICATION ROLE my_app_role;
-                    GRANT USAGE ON STREAMLIT v_schema.my_streamlit TO APPLICATION ROLE my_app_role;
-                    """
-                )
+                f"""
+-- AUTO GENERATED CHILDREN SECTION
+CREATE OR REPLACE STREAMLIT IDENTIFIER('v_schema.my_streamlit')
+FROM = '{custom_dir_path}'
+MAIN_FILE = 'streamlit_app.py';
+CREATE APPLICATION ROLE IF NOT EXISTS my_app_role;
+GRANT USAGE ON SCHEMA v_schema TO APPLICATION ROLE my_app_role;
+GRANT USAGE ON STREAMLIT v_schema.my_streamlit TO APPLICATION ROLE my_app_role;
+"""
             )