SNOW-2223110-asyncjob-failure-status-apis (#3605)

Author: sfc-gh-mayliu

CHANGELOG.md

@@ -7,6 +7,9 @@
 #### New Features
 
 - `Session.create_dataframe` now accepts keyword arguments that are forwarded to the internal call to `Session.write_pandas` or `Session.write_arrow` when creating a DataFrame from a pandas DataFrame or a pyarrow Table.
+- Added new APIs for `AsyncJob`:
+  - `AsyncJob.is_failed()` returns a `bool` indicating if a job has failed. Can be used in combination with `AsyncJob.is_done()` to determine if a job is finished and errored.
+  - `AsyncJob.status()` returns a string representing the current query status (e.g., "RUNNING", "SUCCESS", "FAILED_WITH_ERROR") for detailed monitoring without calling `result()`.
 - Added a dataframe profiler. To use, you can call get_execution_profile() on your desired dataframe. This profiler reports the queries executed to evaluate a dataframe, and statistics about each of the query operators. Currently an experimental feature
 
 ## 1.35.0 (2025-07-24)

src/snowflake/snowpark/async_job.py

@@ -240,6 +240,22 @@ def is_done(self) -> bool:
         is_running = self._session._conn._conn.is_still_running(status)
         return not is_running
 
+    def is_failed(self) -> bool:
+        """
+        Checks the status of the query associated with this instance and returns a bool value
+        indicating whether the query has failed.
+        """
+        status = self._session._conn._conn.get_query_status(self.query_id)
+        return self._session._conn._conn.is_an_error(status)
+
+    def status(self) -> str:
+        """
+        Returns a string representing the current status of the query.
+        (e.g., "RUNNING", "SUCCESS", "FAILED_WITH_ERROR", "ABORTING", etc.)
+        """
+        status = self._session._conn._conn.get_query_status(self.query_id)
+        return status.name
+
     def cancel(self) -> None:
         """Cancels the query associated with this instance."""
         # stop and cancel current query id

tests/integ/scala/test_async_job_suite.py

@@ -378,17 +378,27 @@ def test_async_is_running_and_cancel(session):
     while not async_job.is_done():
         sleep(1.0)
     assert async_job.is_done()
+    assert not async_job.is_failed()
 
     # set 20s to avoid flakiness
     async_job2 = session.sql("select SYSTEM$WAIT(20)").collect_nowait()
     assert not async_job2.is_done()
+    assert not async_job2.is_failed()
     async_job2.cancel()
     start = time()
     while not async_job2.is_done():
+        assert async_job2.status() in [
+            "RUNNING",
+            "ABORTING",
+            "ABORTED",
+            "FAILED_WITH_ERROR",
+        ]
         sleep(1.0)
     # If query is canceled, it takes less time than originally needed
     assert (time() - start) < 20
     assert async_job2.is_done()
+    # cancel/abort should fail the query
+    assert async_job2.is_failed()
 
 
 @pytest.mark.skipif(not is_pandas_available, reason="pandas is not available")
@@ -514,3 +524,44 @@ def test_iter_cursor_wait_for_result(session, action):
     df = session.sql("call system$wait(5)")
     async_job = action(df)
     assert async_job.result() is not None
+
+
+@pytest.mark.skipif(IS_IN_STORED_PROC, reason="sproc is not supported in async job yet")
+def test_async_job_status_apis(session):
+    successful_queries = [
+        "select a from values (1, 2), (3, 4) as t(a, b)",
+        "select 1 + 1",
+    ]
+    for query in successful_queries:
+        async_job = session.sql(query).collect_nowait()
+        while not async_job.is_done():
+            assert not async_job.is_failed()
+            status = async_job.status()
+            assert status in [
+                "RUNNING",
+                "QUEUED",
+                "RESUMING_WAREHOUSE",
+                "QUEUED_REPARING_WAREHOUSE",
+                "SUCCESS",
+            ]
+            sleep(1.0)
+        assert not async_job.is_failed()
+        assert async_job.status() == "SUCCESS"
+        assert async_job.result() is not None
+
+    failed_queries = ["select c from values (1, 2), (3, 4) as t(a, b)", "select 1 / 0"]
+    for query in failed_queries:
+        async_job = session.sql(query).collect_nowait()
+        while not async_job.is_done():
+            sleep(1.0)
+            status = async_job.status()
+            assert status in [
+                "RUNNING",
+                "QUEUED",
+                "RESUMING_WAREHOUSE",
+                "QUEUED_REPARING_WAREHOUSE",
+                "FAILED_WITH_ERROR",
+                "FAILED_WITH_INCIDENT",
+            ]
+        assert async_job.is_failed()
+        assert async_job.status() in ["FAILED_WITH_ERROR", "FAILED_WITH_INCIDENT"]