Databricks: add more methods to represent run state information

airflow/providers/databricks/hooks/databricks.py

@@ -407,6 +407,14 @@ def get_run_state(self, run_id: str) -> RunState:
         """
         Retrieves run state of the run.
 
+        Please note that any Airflow tasks that call the ``get_run_state`` method will result in
+        failure unless you have enabled xcom pickling.  This can be done using the following
+        environment variable: ``AIRFLOW__CORE__ENABLE_XCOM_PICKLING``
+
+        If you do not want to enable xcom pickling, use the ``get_run_state_str`` method to get
+        a string describing state, or ``get_run_state_lifecycle``, ``get_run_state_result``, or
+        ``get_run_state_message`` to get individual components of the run state.
+
         :param run_id: id of the run
         :return: state of the run
         """
@@ -419,6 +427,46 @@ def get_run_state(self, run_id: str) -> RunState:
         state_message = state['state_message']
         return RunState(life_cycle_state, result_state, state_message)
 
+    def get_run_state_str(self, run_id: str) -> str:
+        """
+        Return the string representation of RunState.
+
+        :param run_id: id of the run
+        :return: string describing run state
+        """
+        state = self.get_run_state(run_id)
+        run_state_str = (
+            f"State: {state.life_cycle_state}. Result: {state.result_state}. {state.state_message}"
+        )
+        return run_state_str
+
+    def get_run_state_lifecycle(self, run_id: str) -> str:
+        """
+        Returns the lifecycle state of the run
+
+        :param run_id: id of the run
+        :return: string with lifecycle state
+        """
+        return self.get_run_state(run_id).life_cycle_state
+
+    def get_run_state_result(self, run_id: str) -> str:
+        """
+        Returns the resulting state of the run
+
+        :param run_id: id of the run
+        :return: string with resulting state
+        """
+        return self.get_run_state(run_id).result_state
+
+    def get_run_state_message(self, run_id: str) -> str:
+        """
+        Returns the state message for the run
+
+        :param run_id: id of the run
+        :return: string with state message
+        """
+        return self.get_run_state(run_id).state_message
+
     def cancel_run(self, run_id: str) -> None:
         """
         Cancels the run.

tests/providers/databricks/hooks/test_databricks.py

@@ -389,6 +389,30 @@ def test_get_run_state(self, mock_requests):
             timeout=self.hook.timeout_seconds,
         )
 
+    @mock.patch('airflow.providers.databricks.hooks.databricks.requests')
+    def test_get_run_state_str(self, mock_requests):
+        mock_requests.get.return_value.json.return_value = GET_RUN_RESPONSE
+        run_state_str = self.hook.get_run_state_str(RUN_ID)
+        assert run_state_str == f"State: {LIFE_CYCLE_STATE}. Result: {RESULT_STATE}. {STATE_MESSAGE}"
+
+    @mock.patch('airflow.providers.databricks.hooks.databricks.requests')
+    def test_get_run_state_lifecycle(self, mock_requests):
+        mock_requests.get.return_value.json.return_value = GET_RUN_RESPONSE
+        lifecycle_state = self.hook.get_run_state_lifecycle(RUN_ID)
+        assert lifecycle_state == LIFE_CYCLE_STATE
+
+    @mock.patch('airflow.providers.databricks.hooks.databricks.requests')
+    def test_get_run_state_result(self, mock_requests):
+        mock_requests.get.return_value.json.return_value = GET_RUN_RESPONSE
+        result_state = self.hook.get_run_state_result(RUN_ID)
+        assert result_state == RESULT_STATE
+
+    @mock.patch('airflow.providers.databricks.hooks.databricks.requests')
+    def test_get_run_state_cycle(self, mock_requests):
+        mock_requests.get.return_value.json.return_value = GET_RUN_RESPONSE
+        state_message = self.hook.get_run_state_message(RUN_ID)
+        assert state_message == STATE_MESSAGE
+
     @mock.patch('airflow.providers.databricks.hooks.databricks.requests')
     def test_cancel_run(self, mock_requests):
         mock_requests.post.return_value.json.return_value = GET_RUN_RESPONSE