class_docstr: Hook for Google DataFusion.
class_docstr_tok: ["Hook","for","Google","DataFusion","."]
code: def get_pipeline_workflow(
      self,
      pipeline_name: str,
      instance_url: str,
      pipeline_id: str,
      pipeline_type: DataFusionPipelineType = DataFusionPipelineType.BATCH,
      namespace: str = "default",
  ) -> Any:
      url = os.path.join(
          self._base_url(instance_url, namespace),
          quote(pipeline_name),
          f"{self.cdap_program_type(pipeline_type=pipeline_type)}s",
          self.cdap_program_id(pipeline_type=pipeline_type),
          "runs",
          quote(pipeline_id),
      )
      response = self._cdap_request(url=url, method="GET")
      self._check_response_status_and_data(
          response,
          f"Retrieving a pipeline state failed with code {response.status}",
      )
      workflow = json.loads(response.data)
      return workflow
  
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 19 for calling the DataFusionHook obj's underlying member method code with example usage: obj.get_pipeline_workflow(pipeline_name, instance_url, pipeline_id, pipeline_type, namespace) and returns: workflow