Add stage type property to build-image

src/snowflake/cli/_plugins/spcs/services/commands.py

@@ -35,11 +35,13 @@
 from snowflake.cli._plugins.spcs.services.service_project_paths import (
     ServiceProjectPaths,
 )
-from snowflake.cli._plugins.stage.manager import StageManager
+from snowflake.cli._plugins.stage.manager import (
+    InternalStageEncryptionType,
+    StageManager,
+)
 from snowflake.cli.api.cli_global_context import get_cli_context
 from snowflake.cli.api.commands.decorators import with_project_definition
 from snowflake.cli.api.commands.flags import (
-    IfExistsOption,
     IfNotExistsOption,
     OverrideableOption,
     entity_argument,
@@ -178,32 +180,9 @@ def _service_name_callback(name: FQN) -> FQN:
         help_example='`list --like "my%"` lists all services that begin with “my”.'
     ),
     scope_option=scope_option(help_example="`list --in compute-pool my_pool`"),
-    ommit_commands=["drop"],
 )
 
 
-@app.command(requires_connection=True)
-def drop(
-    name: FQN = ServiceNameArgument,
-    if_exists: bool = IfExistsOption(),
-    force: bool = typer.Option(
-        False,
-        "--force",
-        help="Drops the service along with any block storage volumes it contains. Without this flag, dropping a service that contains block storage volumes fails.",
-        is_flag=True,
-    ),
-    **options,
-) -> CommandResult:
-    """
-    Drops a service.
-    """
-    return SingleQueryResult(
-        ServiceManager().drop(
-            service_name=name.sql_identifier, if_exists=if_exists, force=force
-        )
-    )
-
-
 @app.command(requires_connection=True)
 def create(
     name: FQN = ServiceNameArgument,
@@ -707,6 +686,15 @@ def build_image(
         help="Stage to store build context files. Format: [db.][schema.]stage_name. If not provided, a temporary stage will be created and dropped automatically. If provided, only the uploaded build context files will be removed after the build completes.",
         show_default=False,
     ),
+    stage_encryption: Optional[str] = typer.Option(
+        None,
+        "--stage-encryption",
+        help=(
+            "SNOWFLAKE_SSE or SNOWFLAKE_FULL for an auto-created temporary stage; "
+            "ignored when --stage is set. Omit for legacy CREATE STAGE (no ENCRYPTION clause)."
+        ),
+        show_default=False,
+    ),
     job_name: str = typer.Option(
         None,
         "--job-name",
@@ -741,6 +729,9 @@ def build_image(
     If --stage is not provided, a stage will be automatically created using the
     current session's database and schema context, and dropped after the build completes.
     If your session doesn't have a database/schema set, you should provide --stage explicitly.
+
+    Optional --stage-encryption applies only to that auto-created stage; use SNOWFLAKE_SSE
+    when stage-mounted builds require server-side encryption on your deployment.
     """
     # Verify Dockerfile exists in build context directory
     dockerfile_path = build_context_dir / "Dockerfile"
@@ -776,15 +767,31 @@ def build_image(
                 f"Invalid job name '{job_name}'. Must be a valid unquoted identifier."
             )
 
-    stage_manager = StageManager()
     use_temporary_stage = stage is None
 
+    temp_encryption: InternalStageEncryptionType | None = None
+    if stage_encryption is not None:
+        key = stage_encryption.strip().upper()
+        try:
+            temp_encryption = InternalStageEncryptionType(key)
+        except ValueError:
+            allowed = ", ".join(sorted(e.value for e in InternalStageEncryptionType))
+            raise CliArgumentError(
+                f"Invalid --stage-encryption {stage_encryption!r}. "
+                f"Expected one of: {allowed}."
+            )
+
+    stage_manager = StageManager()
+
     if use_temporary_stage:
         # Create a stage
         stage = f"{job_name}_stage"
         cli_console.step(f"Creating temporary stage: {stage}")
         stage_fqn = FQN.from_string(stage).using_context()
-        stage_manager.create(fqn=stage_fqn)
+        if temp_encryption is not None:
+            stage_manager.create(fqn=stage_fqn, encryption=temp_encryption)
+        else:
+            stage_manager.create(fqn=stage_fqn)
     else:
         # Use the provided stage (ensure it exists)
         stage_fqn = FQN.from_string(stage)

src/snowflake/cli/_plugins/spcs/services/manager.py

@@ -304,50 +304,26 @@ def execute_job(
             replicas=replicas,
         )
 
-    def build_image(
+    def build_image_registry_url(
         self,
-        job_service_name: str,
-        compute_pool: str,
         image_repository: str,
-        image_name: str,
-        image_tag: str,
-        stage: str,
-        build_context_path: str,
-        external_access_integrations: Optional[List[str]],
-        async_mode: bool = True,
-    ) -> SnowflakeCursor:
-        """
-        Builds a container image using EXECUTE JOB SERVICE with the Snowflake image builder.
+    ) -> str:
+        """Build the canonical image-registry URL for ``image_repository``.
 
-        Args:
-            job_service_name: Name for the build job service
-            compute_pool: Compute pool to run the build job
-            image_repository: Repository path in format [db.][schema.]repo_name
-            image_name: Name for the built image
-            image_tag: Tag for the built image
-            stage: Stage where the build context is uploaded
-            build_context_path: Path on stage where build context is located
-            external_access_integrations: List of EAI names
-            async_mode: Whether to run the build asynchronously
+        Returns ``<org>-<account>.registry-local.snowflakecomputing.com/<db>/<schema>/<repo>``,
+        the same shape ``build_image`` injects as the ``IMAGE_REGISTRY_URL`` env var. Exposed
+        separately so the post-push validation hook can build the URL it passes to
+        ``SYSTEM$START_IMAGE_VALIDATION_SERVICE`` without re-implementing FQN parsing.
+
+        Raises ``ValueError`` if ``image_repository`` is missing the database or schema parts.
         """
-        # Get organization name
-        # Using execute_string (same as get_account) for consistency
         *_, org_cursor = self._conn.execute_string(
             "SELECT CURRENT_ORGANIZATION_NAME()", cursor_class=DictCursor
         )
         org_name = org_cursor.fetchone()["CURRENT_ORGANIZATION_NAME()"].lower()
-
-        # Get account name using existing utility function
         account_name = get_account(self._conn)
 
-        # Parse the image repository using FQN utility
-        # This handles [db.][schema.]repo_name format automatically
         repo_fqn = FQN.from_string(image_repository).using_connection(self._conn)
-
-        # Validate that we have all required parts
-        # Note: FQN parsing gives us: "db.schema.repo" → both present,
-        # "schema.repo" → only schema, "repo" → neither
-        # It's impossible to have database without schema
         if not repo_fqn.database or not repo_fqn.schema:
             missing_parts = []
             if not repo_fqn.database:
@@ -362,26 +338,53 @@ def build_image(
                 f"Provided: '{image_repository}'"
             )
 
-        # Build the image registry URL
-        # Format: <org>-<account-name>.registry-local.snowflakecomputing.com/<db>/<schema>/<repo>
-        image_registry_url = (
+        return (
             f"{org_name}-{account_name}.registry-local.snowflakecomputing.com/"
             f"{repo_fqn.database.lower()}/{repo_fqn.schema.lower()}/{repo_fqn.name.lower()}"
         )
 
-        # Create the specification for the image build job
+    def build_image(
+        self,
+        job_service_name: str,
+        compute_pool: str,
+        image_repository: str,
+        image_name: str,
+        image_tag: str,
+        stage: str,
+        build_context_path: str,
+        external_access_integrations: Optional[List[str]],
+        async_mode: bool = True,
+    ) -> SnowflakeCursor:
+        """
+        Builds a container image using EXECUTE JOB SERVICE with the Snowflake image builder.
+
+        Args:
+            job_service_name: Name for the build job service
+            compute_pool: Compute pool to run the build job
+            image_repository: Repository path in format [db.][schema.]repo_name
+            image_name: Name for the built image
+            image_tag: Tag for the built image
+            stage: Stage where the build context is uploaded
+            build_context_path: Path on stage where build context is located
+            external_access_integrations: List of EAI names
+            async_mode: Whether to run the build asynchronously
+        """
+        image_registry_url = self.build_image_registry_url(image_repository)
+
+        env_vars = {
+            "IMAGE_REGISTRY_URL": image_registry_url,
+            "IMAGE_NAME": image_name,
+            "IMAGE_TAG": image_tag,
+            "BUILD_CONTEXT": "/app",
+        }
+
         spec = {
             "spec": {
                 "containers": [
                     {
                         "name": "main",
                         "image": "/snowflake/images/snowflake_images/sf-image-build:0.0.1",
-                        "env": {
-                            "IMAGE_REGISTRY_URL": image_registry_url,
-                            "IMAGE_NAME": image_name,
-                            "IMAGE_TAG": image_tag,
-                            "BUILD_CONTEXT": "/app",
-                        },
+                        "env": env_vars,
                         "volumeMounts": [
                             {
                                 "name": "code-volume",
@@ -711,15 +714,6 @@ def suspend(self, service_name: str):
     def resume(self, service_name: str):
         return self.execute_query(f"alter service {service_name} resume")
 
-    def drop(
-        self, service_name: str, if_exists: bool = False, force: bool = False
-    ) -> SnowflakeCursor:
-        if_exists_clause = " if exists" if if_exists else ""
-        force_clause = " force" if force else ""
-        return self.execute_query(
-            f"drop service{if_exists_clause} {service_name}{force_clause}"
-        )
-
     def set_property(
         self,
         service_name: str,

tests/__snapshots__/test_help_messages.ambr

@@ -18436,6 +18436,11 @@
   |                                            the uploaded build context files  |
   |                                            will be removed after the build   |
   |                                            completes.                        |
+  |    --stage-encryption         TEXT       SNOWFLAKE_SSE or SNOWFLAKE_FULL for an |
+  |                                            auto-created temporary stage;      |
+  |                                            ignored when --stage is set. Omit for|
+  |                                            legacy CREATE STAGE (no ENCRYPTION |
+  |                                            clause).                           |
   |    --job-name                   TEXT       Name for the build job service.   |
   |                                            If not provided, a name will be   |
   |                                            auto-generated.                   |