class_docstr: Runs a Data Pipelines Instance using the Data Pipelines API.
  
  :param data_pipeline_name:  The display name of the pipeline. In example
      projects/PROJECT_ID/locations/LOCATION_ID/pipelines/PIPELINE_ID it would be the PIPELINE_ID.
  :param project_id: The ID of the GCP project that owns the job.
  :param location: The location to direct the Data Pipelines instance to (for example us-central1).
  :param gcp_conn_id: The connection ID to connect to the Google Cloud
      Platform.
  
  Returns the created Job in JSON representation.
class_docstr_tok: ["Runs","a","Data","Pipelines","Instance","using","the","Data","Pipelines","API",".",":","param","data_pipeline_name",":","The","display","name","of","the","pipeline",".","In","example","projects\/PROJECT_ID\/locations\/LOCATION_ID\/pipelines\/PIPELINE_ID","it","would","be","the","PIPELINE_ID",".",":","param","project_id",":","The","ID","of","the","GCP","project","that","owns","the","job",".",":","param","location",":","The","location","to","direct","the","Data","Pipelines","instance","to","(","for","example","us-central1",")",".",":","param","gcp_conn_id",":","The","connection","ID","to","connect","to","the","Google","Cloud","Platform",".","Returns","the","created","Job","in","JSON","representation","."]
code: def execute(self, context: Context):
      self.data_pipeline_hook = DataPipelineHook(
          gcp_conn_id=self.gcp_conn_id
      )
  
      if self.data_pipeline_name is None:
          raise AirflowException(
              "Data Pipeline name not given; cannot run unspecified pipeline."
          )
      if self.project_id is None:
          raise AirflowException(
              "Data Pipeline Project ID not given; cannot run pipeline."
          )
      if self.location is None:
          raise AirflowException(
              "Data Pipeline location not given; cannot run pipeline."
          )
  
      self.response = self.data_pipeline_hook.run_data_pipeline(
          data_pipeline_name=self.data_pipeline_name,
          project_id=self.project_id,
          location=self.location,
      )
  
      if self.response:
          if "error" in self.response:
              raise AirflowException(
                  self.response.get("error").get("message")
              )
  
      return self.response
  
file_path: airflow/airflow/providers/google/cloud/operators/datapipeline.py
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Sequence
  from airflow.exceptions import AirflowException
  from airflow.providers.google.cloud.hooks.datapipeline import DEFAULT_DATAPIPELINE_LOCATION, DataPipelineHook
  from airflow.providers.google.cloud.operators.cloud_base import GoogleCloudBaseOperator
label_desc: Use image node_id 2 for calling the RunDataPipelineOperator obj's underlying member method code with example usage: obj.execute(context) and returns: self