Source code for airflow.providers.amazon.aws.sensors.rds
# 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.abcimportSequencefromfunctoolsimportcached_propertyfromtypingimportTYPE_CHECKINGfromairflow.exceptionsimportAirflowNotFoundExceptionfromairflow.providers.amazon.aws.hooks.rdsimportRdsHookfromairflow.providers.amazon.aws.utils.rdsimportRdsDbTypefromairflow.sensors.baseimportBaseSensorOperatorifTYPE_CHECKING:fromairflow.utils.contextimportContextclassRdsBaseSensor(BaseSensorOperator):"""Base operator that implements common functions for all sensors."""ui_color="#ddbb77"ui_fgcolor="#ffffff"def__init__(self,*args,aws_conn_id:str|None="aws_conn_id",hook_params:dict|None=None,**kwargs):self.hook_params=hook_paramsor{}self.aws_conn_id=aws_conn_idself.target_statuses:list[str]=[]super().__init__(*args,**kwargs)@cached_propertydefhook(self):returnRdsHook(aws_conn_id=self.aws_conn_id,**self.hook_params)
[docs]classRdsSnapshotExistenceSensor(RdsBaseSensor):""" Waits for RDS snapshot with a specific status. .. seealso:: For more information on how to use this sensor, take a look at the guide: :ref:`howto/sensor:RdsSnapshotExistenceSensor` :param db_type: Type of the DB - either "instance" or "cluster" :param db_snapshot_identifier: The identifier for the DB snapshot :param target_statuses: Target status of snapshot """
[docs]defpoke(self,context:Context):self.log.info("Poking for statuses : %s\nfor snapshot %s",self.target_statuses,self.db_snapshot_identifier)try:ifself.db_type.value=="instance":state=self.hook.get_db_snapshot_state(self.db_snapshot_identifier)else:state=self.hook.get_db_cluster_snapshot_state(self.db_snapshot_identifier)exceptAirflowNotFoundException:returnFalsereturnstateinself.target_statuses
[docs]classRdsExportTaskExistenceSensor(RdsBaseSensor):""" Waits for RDS export task with a specific status. .. seealso:: For more information on how to use this sensor, take a look at the guide: :ref:`howto/sensor:RdsExportTaskExistenceSensor` :param export_task_identifier: A unique identifier for the snapshot export task. :param target_statuses: Target status of export task """
[docs]defpoke(self,context:Context):self.log.info("Poking for statuses : %s\nfor export task %s",self.target_statuses,self.export_task_identifier)try:state=self.hook.get_export_task_state(self.export_task_identifier)exceptAirflowNotFoundException:returnFalsereturnstateinself.target_statuses
[docs]classRdsDbSensor(RdsBaseSensor):""" Waits for an RDS instance or cluster to enter one of a number of states. .. seealso:: For more information on how to use this sensor, take a look at the guide: :ref:`howto/sensor:RdsDbSensor` :param db_type: Type of the DB - either "instance" or "cluster" (default: 'instance') :param db_identifier: The AWS identifier for the DB :param target_statuses: Target status of DB """
[docs]defpoke(self,context:Context):db_type=RdsDbType(self.db_type)self.log.info("Poking for statuses : %s\nfor db instance %s",self.target_statuses,self.db_identifier)try:ifdb_type==RdsDbType.INSTANCE:state=self.hook.get_db_instance_state(self.db_identifier)else:state=self.hook.get_db_cluster_state(self.db_identifier)exceptAirflowNotFoundException:returnFalsereturnstateinself.target_statuses