class_docstr: Hook for Google Dataflow.
  
  All the methods in the hook where project_id is used must be called with
  keyword arguments rather than positional.
class_docstr_tok: ["Hook","for","Google","Dataflow",".","All","the","methods","in","the","hook","where","project_id","is","used","must","be","called","with","keyword","arguments","rather","than","positional","."]
docstr: Starts flex templates with the Dataflow pipeline.
  
  :param body: The request body. See:
      https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.locations.flexTemplates/launch#request-body
  :param location: The location of the Dataflow job (for example europe-west1)
  :param project_id: The ID of the GCP project that owns the job.
      If set to ``None`` or missing, the default project_id from the GCP connection is used.
  :param on_new_job_id_callback: (Deprecated) A callback that is called when a Job ID is detected.
  :param on_new_job_callback: A callback that is called when a Job is detected.
  :return: the Job
code: def start_flex_template(
      self,
      body: dict,
      location: str,
      project_id: str,
      on_new_job_id_callback: Callable[[str], None] | None = None,
      on_new_job_callback: Callable[[dict], None] | None = None,
  ) -> dict:
      """
      Starts flex templates with the Dataflow pipeline.
  
      :param body: The request body. See:
          https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.locations.flexTemplates/launch#request-body
      :param location: The location of the Dataflow job (for example europe-west1)
      :param project_id: The ID of the GCP project that owns the job.
          If set to ``None`` or missing, the default project_id from the GCP connection is used.
      :param on_new_job_id_callback: (Deprecated) A callback that is called when a Job ID is detected.
      :param on_new_job_callback: A callback that is called when a Job is detected.
      :return: the Job
      """
      service = self.get_conn()
      request = (
          service.projects()
          .locations()
          .flexTemplates()
          .launch(projectId=project_id, body=body, location=location)
      )
      response = request.execute(num_retries=self.num_retries)
      job = response["job"]
  
      if on_new_job_id_callback:
          warnings.warn(
              "on_new_job_id_callback is Deprecated. Please start using on_new_job_callback",
              AirflowProviderDeprecationWarning,
              stacklevel=3,
          )
          on_new_job_id_callback(job.get("id"))
  
      if on_new_job_callback:
          on_new_job_callback(job)
  
      jobs_controller = _DataflowJobsController(
          dataflow=self.get_conn(),
          project_number=project_id,
          job_id=job.get("id"),
          location=location,
          poll_sleep=self.poll_sleep,
          num_retries=self.num_retries,
          cancel_timeout=self.cancel_timeout,
          wait_until_finished=self.wait_until_finished,
      )
      jobs_controller.wait_for_done()
  
      return jobs_controller.get_jobs(refresh=True)[0]
  
imports: from __future__ import annotations
  import functools
  import json
  import re
  import shlex
  import subprocess
  import time
  import uuid
  import warnings
  from copy import deepcopy
  from typing import TYPE_CHECKING, Any, Callable, Generator, Sequence, TypeVar, cast
  from google.cloud.dataflow_v1beta3 import GetJobRequest, Job, JobState, JobsV1Beta3AsyncClient, JobView
  from google.cloud.dataflow_v1beta3.types.jobs import ListJobsRequest
  from googleapiclient.discovery import build
  from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
  from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType, beam_options_to_args
  from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID, GoogleBaseAsyncHook, GoogleBaseHook
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.timeout import timeout
label_desc: Use image node_id 6 for calling the DataflowHook obj's underlying member method code with example usage: obj.start_flex_template(body, location, project_id, on_new_job_id_callback, on_new_job_callback) and returns: jobs_controller