Source code for airflow.providers.google.cloud.sensors.looker

#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""This module contains Google Cloud Looker sensors."""
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

if TYPE_CHECKING:
    from airflow.utils.context import Context


[docs]class LookerCheckPdtBuildSensor(BaseSensorOperator): """ 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. """
[docs] template_fields = ["materialization_id"]
def __init__( self, materialization_id: str, looker_conn_id: str, cancel_on_kill: bool = True, **kwargs ) -> None: super().__init__(**kwargs) self.materialization_id = materialization_id self.looker_conn_id = looker_conn_id self.cancel_on_kill = cancel_on_kill self.hook: LookerHook | None = None
[docs] 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
[docs] def on_kill(self): if self.materialization_id and self.cancel_on_kill: self.hook.stop_pdt_build(materialization_id=self.materialization_id)

Was this entry helpful?