Source code for airflow.providers.amazon.aws.sensors.redshift_cluster
# 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.from__future__importannotationsfromcollections.abcimportSequencefromdatetimeimporttimedeltafromfunctoolsimportcached_propertyfromtypingimportTYPE_CHECKING,Anyfromairflow.configurationimportconffromairflow.exceptionsimportAirflowExceptionfromairflow.providers.amazon.aws.hooks.redshift_clusterimportRedshiftHookfromairflow.providers.amazon.aws.triggers.redshift_clusterimportRedshiftClusterTriggerfromairflow.providers.amazon.aws.utilsimportvalidate_execute_complete_eventfromairflow.sensors.baseimportBaseSensorOperatorifTYPE_CHECKING:fromairflow.utils.contextimportContext
[docs]classRedshiftClusterSensor(BaseSensorOperator):""" Waits for a Redshift cluster to reach a specific status. .. seealso:: For more information on how to use this sensor, take a look at the guide: :ref:`howto/sensor:RedshiftClusterSensor` :param cluster_identifier: The identifier for the cluster being pinged. :param target_status: The cluster status desired. :param deferrable: Run operator in the deferrable mode. """
[docs]defpoke(self,context:Context)->bool:current_status=self.hook.cluster_status(self.cluster_identifier)self.log.info("Poked cluster %s for status '%s', found status '%s'",self.cluster_identifier,self.target_status,current_status,)returncurrent_status==self.target_status
[docs]defexecute_complete(self,context:Context,event:dict[str,Any]|None=None)->None:validated_event=validate_execute_complete_event(event)status=validated_event["status"]ifstatus=="error":raiseAirflowException(f"{validated_event['status']}: {validated_event['message']}")elifstatus=="success":self.log.info("%s completed successfully.",self.task_id)self.log.info("Cluster Identifier %s is in %s state",self.cluster_identifier,self.target_status)