class_docstr: Hook for Google DataFusion.
class_docstr_tok: ["Hook","for","Google","DataFusion","."]
docstr: Polls pipeline state and raises an exception if the state fails or times out.
docstr_tok: ["Polls","pipeline","state","and","raises","an","exception","if","the","state","fails","or","times","out","."]
code: def wait_for_pipeline_state(
      self,
      pipeline_name: str,
      pipeline_id: str,
      instance_url: str,
      pipeline_type: DataFusionPipelineType = DataFusionPipelineType.BATCH,
      namespace: str = "default",
      success_states: list[str] | None = None,
      failure_states: list[str] | None = None,
      timeout: int = 5 * 60,
  ) -> None:
      """Polls pipeline state and raises an exception if the state fails or times out."""
      failure_states = failure_states or FAILURE_STATES
      success_states = success_states or SUCCESS_STATES
      start_time = time.monotonic()
      current_state = None
      while time.monotonic() - start_time < timeout:
          try:
              workflow = self.get_pipeline_workflow(
                  pipeline_name=pipeline_name,
                  pipeline_id=pipeline_id,
                  pipeline_type=pipeline_type,
                  instance_url=instance_url,
                  namespace=namespace,
              )
              current_state = workflow["status"]
          except AirflowException:
              pass  # Because the pipeline may not be visible in system yet
          if current_state in success_states:
              return
          if current_state in failure_states:
              raise AirflowException(
                  f"Pipeline {pipeline_name} state {current_state} is not one of {success_states}"
              )
          time.sleep(30)
  
      # Time is up!
      raise AirflowException(
          f"Pipeline {pipeline_name} state {current_state} is not "
          f"one of {success_states} after {timeout}s"
      )
  
imports: from __future__ import annotations
  import asyncio
  import json
  import os
  import time
  from typing import Any, Dict, Sequence
  from urllib.parse import quote, urlencode, urljoin
  import google.auth
  from aiohttp import ClientSession
  from gcloud.aio.auth import AioSession, Token
  from google.api_core.retry import exponential_sleep_generator
  from googleapiclient.discovery import Resource, build
  from airflow.exceptions import AirflowException, AirflowNotFoundException
  from airflow.providers.google.cloud.utils.datafusion import DataFusionPipelineType
  from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID, GoogleBaseAsyncHook, GoogleBaseHook
label_desc: Use image node_id 3 for calling the DataFusionHook obj's underlying member method code with example usage: obj.wait_for_pipeline_state(pipeline_name, pipeline_id, instance_url, pipeline_type, namespace, success_states, failure_states, timeout) without return types