Added async option to spcs service execute-job command

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

@@ -144,6 +144,13 @@ def _service_name_callback(name: FQN) -> FQN:
     help=_QUERY_WAREHOUSE_HELP,
 )
 
+_ASYNC_HELP = "[TRUE | FALSE] Specifies whether to execute the job asynchronously."
+AsyncOption = OverrideableOption(
+    "FALSE",
+    "--async",
+    help=_ASYNC_HELP,
+)
+
 _AUTO_RESUME_HELP = "The service will automatically resume when a service function or ingress is called."
 AutoResumeOption = OverrideableOption(
     True,
@@ -161,6 +168,7 @@ def _service_name_callback(name: FQN) -> FQN:
 
 _COMMENT_HELP = "Comment for the service."
 
+
 add_object_command_aliases(
     app=app,
     object_type=ObjectType.SERVICE,
@@ -277,6 +285,7 @@ def execute_job(
     ),
     query_warehouse: Optional[str] = QueryWarehouseOption(),
     comment: Optional[str] = CommentOption(help=_COMMENT_HELP),
+    async_execution: Optional[str] = AsyncOption(),
     **options,
 ) -> CommandResult:
     """
@@ -289,6 +298,7 @@ def execute_job(
         external_access_integrations=external_access_integrations,
         query_warehouse=query_warehouse,
         comment=comment,
+        async_execution=async_execution,
     )
     return SingleQueryResult(cursor)
 

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

@@ -226,11 +226,13 @@ def execute_job(
         external_access_integrations: Optional[List[str]],
         query_warehouse: Optional[str],
         comment: Optional[str],
+        async_execution: Optional[str]
     ) -> SnowflakeCursor:
         spec = self._read_yaml(spec_path)
         query = f"""\
                 EXECUTE JOB SERVICE
                 IN COMPUTE POOL {compute_pool}
+                ASYNC = {async_execution}
                 FROM SPECIFICATION $$
                 {spec}
                 $$

tests/__snapshots__/test_help_messages.ambr

@@ -16930,6 +16930,9 @@
   |                                     without explicitly specifying a          |
   |                                     warehouse to use.                        |
   |    --comment                  TEXT  Comment for the service.                 |
+  |    --async                    TEXT  [TRUE | FALSE] Specifies whether to      |
+  |                                     execute the job asynchronously.          |
+  |                                     [default: FALSE]                         |
   |    --help             -h            Show this message and exit.              |
   +------------------------------------------------------------------------------+
   +- Connection configuration ---------------------------------------------------+

tests/spcs/test_services.py

@@ -582,6 +582,7 @@ def test_execute_job_service(mock_execute_query, temporary_directory):
     ]
     query_warehouse = "test_warehouse"
     comment = "'user\\'s comment'"
+    async_execution = "FALSE"
 
     cursor = Mock(spec=SnowflakeCursor)
     mock_execute_query.return_value = cursor
@@ -593,11 +594,13 @@ def test_execute_job_service(mock_execute_query, temporary_directory):
         external_access_integrations=external_access_integrations,
         query_warehouse=query_warehouse,
         comment=comment,
+        async_execution=async_execution,
     )
     expected_query = " ".join(
         [
             "EXECUTE JOB SERVICE",
             "IN COMPUTE POOL test_pool",
+            "ASYNC = FALSE",
             f"FROM SPECIFICATION $$ {json.dumps(SPEC_DICT)} $$",
             "NAME = test_job_service",
             "EXTERNAL_ACCESS_INTEGRATIONS = (google_apis_access_integration,salesforce_api_access_integration)",
@@ -637,6 +640,7 @@ def test_execute_job_service_cli_defaults(
         external_access_integrations=None,
         query_warehouse=None,
         comment=None,
+        async_execution="FALSE",
     )
 
 
@@ -663,6 +667,8 @@ def test_execute_job_service_cli(mock_execute_job, temporary_directory, runner):
             "test_warehouse",
             "--comment",
             "this is a test",
+            "--async",
+            "FALSE"
         ]
     )
     assert result.exit_code == 0, result.output
@@ -673,6 +679,7 @@ def test_execute_job_service_cli(mock_execute_job, temporary_directory, runner):
         external_access_integrations=["google_api", "salesforce_api"],
         query_warehouse="test_warehouse",
         comment=to_string_literal("this is a test"),
+        async_execution="FALSE",
     )
 
 
@@ -681,6 +688,7 @@ def test_execute_job_service_with_invalid_spec(mock_read_yaml):
     job_service_name = "test_job_service"
     compute_pool = "test_pool"
     spec_path = "/path/to/spec.yaml"
+    async_execution = "FALSE"
     external_access_integrations = query_warehouse = comment = None
     mock_read_yaml.side_effect = YAMLError("Invalid YAML")
 
@@ -692,6 +700,7 @@ def test_execute_job_service_with_invalid_spec(mock_read_yaml):
             external_access_integrations=external_access_integrations,
             query_warehouse=query_warehouse,
             comment=comment,
+            async_execution=async_execution,
         )