class_docstr: Check for the state of a previously submitted PDT materialization job.
  
  :param materialization_id: Required. The materialization job ID to poll. (templated)
  :param looker_conn_id: Required. The connection ID to use connecting to Looker.
  :param cancel_on_kill: Optional. Flag which indicates whether cancel the hook's job or not,
      when on_kill is called.
code: def poke(self, context: Context) -> bool:
      self.hook = LookerHook(looker_conn_id=self.looker_conn_id)
  
      if not self.materialization_id:
          # TODO: remove this if check when min_airflow_version is set to higher than 2.7.1
          message = "Invalid `materialization_id`."
          if self.soft_fail:
              raise AirflowSkipException(message)
          raise AirflowException(message)
  
      # materialization_id is templated var pulling output from start task
      status_dict = self.hook.pdt_build_status(
          materialization_id=self.materialization_id
      )
      status = status_dict["status"]
  
      if status == JobStatus.ERROR.value:
          msg = status_dict["message"]
          # TODO: remove this if check when min_airflow_version is set to higher than 2.7.1
          message = f'PDT materialization job failed. Job id: {self.materialization_id}. Message:\n"{msg}"'
          if self.soft_fail:
              raise AirflowSkipException(message)
          raise AirflowException(message)
      elif status == JobStatus.CANCELLED.value:
          # TODO: remove this if check when min_airflow_version is set to higher than 2.7.1
          message = f"PDT materialization job was cancelled. Job id: {self.materialization_id}."
          if self.soft_fail:
              raise AirflowSkipException(message)
          raise AirflowException(message)
      elif status == JobStatus.UNKNOWN.value:
          # TODO: remove this if check when min_airflow_version is set to higher than 2.7.1
          message = f"PDT materialization job has unknown status. Job id: {self.materialization_id}."
          if self.soft_fail:
              raise AirflowSkipException(message)
          raise AirflowException(message)
      elif status == JobStatus.DONE.value:
          self.log.debug(
              "PDT materialization job completed successfully. Job id: %s.",
              self.materialization_id,
          )
          return True
  
      self.log.info(
          "Waiting for PDT materialization job to complete. Job id: %s.",
          self.materialization_id,
      )
      return False
  
file_path: airflow/airflow/providers/google/cloud/sensors/looker.py
imports: from __future__ import annotations
  from typing import TYPE_CHECKING
  from airflow.exceptions import AirflowException, AirflowSkipException
  from airflow.providers.google.cloud.hooks.looker import JobStatus, LookerHook
  from airflow.sensors.base import BaseSensorOperator
label_desc: Use image node_id 2 for calling the LookerCheckPdtBuildSensor obj's underlying member method code with example usage: obj.poke(context) and returns: False, True