Add RdsDbSensor to amazon provider package

[hankehly]

Related: #25952

Summary

This PR adds the RdsDbSensor to the amazon provider package. It waits for an RDS instance or cluster to reach one (or more) of the DB instance states described here.

Todo

• Add RdsDbSensor
• Add/run unit tests

breeze shell
pytest tests/providers/amazon/aws/sensors/test_rds.py::TestRdsDbSensor

• Add/run system tests

breeze shell
export AWS_ACCESS_KEY=***
export AWS_SECRET_ACCESS_KEY=***
airflow dags test -S tests/system/providers/amazon/aws/rds/example_rds_instance.py example_rds_instance 2022-08-01

• Update documentation

[hankehly]

Current sensors rely on the same botocore response structure. The response to describe_db_instances uses a different field name to describe the state of the requested resource. I chose to define the field name as an instance variable because..

• no need to update other sensor classes
• no need to modify method signatures
• minimal code changes
• no extra conditional statements
• it makes sense to define "sensor specific metadata" on the sensor class/instance

[kazanzhy]

I think this variable could be in _check_item method.
Based on item_type you could identify the status field

[hankehly]

You're right, but that requires adding an if-else conditional statement. Is the current approach OK with you?

[kazanzhy]

I think it is OK to add one more if-else.
Actually, I meant something like:

    def _check_item(self, item_type: str, item_name: str) -> bool:
        """Get certain item from `_describe_item()` and check its status"""
        if item_type == 'db_instance':
            status_field = 'DBInstanceStatus'
        else:
            status_field = 'Status'

        try:
            items = self._describe_item(item_type, item_name)
        except ClientError:
            return False
        else:
            return bool(items) and any(
                map(lambda status: items[0][status_field].lower() == status, self.target_statuses)
            )

Seems only DBInstance has this issue with the Status field.
WDYT?

[hankehly]

By adding if/else logic to the base class, we're making it "aware" of the details of the subclass. I think a better approach is to let the subclass tell us what it needs.

I checked the botocore documentation and it looks like "DBInstanceStatus" and "Status" are the only 2 options, so I'll go ahead and make the change to finish up this PR.

[hankehly]

Changed in afefe36

[hankehly]

[hankehly]

2022/09/05 update

[hankehly]

@o-nikolas @vincbeck @ferruzzi @potiuk
Please review this at your earliest convenience.

[vincbeck]

LGTM

[ferruzzi]

Left a couple comments; non-blocking.

[ferruzzi]

My concerns have been addressed, thanks. LGTM

[hankehly]

@o-nikolas @vincbeck @ferruzzi @potiuk
Could someone please merge this PR, or let me know what else needs to change

[o-nikolas]

@o-nikolas @vincbeck @ferruzzi @potiuk Could someone please merge this PR, or let me know what else needs to change

Unfortunately none of us tagged above have the powers to merge code except for @potiuk

[kazanzhy]

Also, the return could be modified from map to the generator if you wish.

Suggested change

	return bool(items) and any(
	map(lambda status: items[0][status_field].lower() == status, self.target_statuses)
	)
	return bool(items) and any(items[0][status_field].lower() == status for status in self.target_statuses)

For me, both variants are well readable but maybe this one is better. WDYT?

[hankehly]

Thanks for the suggestions. I'd like to make these updates in a separate PR.
The formatting is the result of black (line too long)

[hankehly]

Update 2022/09/06
#26003 (comment)

[hankehly]

@o-nikolas @ferruzzi @vincbeck @kazanzhy
I renamed RdsInstanceSensor to RdsDbSensor and added the db_type parameter to match other sensor classes.

I left comments where the code changed.
Please re-OK the changes at your earliest convenience.

[ferruzzi]

LGTM, thanks for the updates.