hankehly commented on issue #25952: URL: https://github.com/apache/airflow/issues/25952#issuecomment-1237585545
cc @dstandish @o-nikolas @ferruzzi @kazanzhy I'd like someone's opinion please. While editing at the operator/sensor code, I'm noticing a couple things: * some duplication in [`RdsBaseSensor`](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/sensors/rds.py) / [`RdsBaseOperator`](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/operators/rds.py) around the `_describe_item` method * base classes are responsible for details of subclasses The current implementation is a great start, but I think we could improve the design to reduce duplication. How about creating (private) reusable models of each RDS resource (ie db instance, cluster, snapshot, etc..) like in the following example? ```py # 1. Create a shared model class for RdsInstance class RdsInstance: def __init__(self, hook, id): self.hook = hook self.id = id def check_status(self, target_statuses: List[str]) -> bool: instances = self.hook.conn.describe_db_instances(DBInstanceIdentifier=self.id) return instances[0]["DbInstanceStatus"] in target_statuses def start(self): pass # todo def stop(self): pass # todo # 2. Create a shared model class for RdsCluster class RdsCluster: def __init__(self, hook, id): self.hook = hook self.id = id def check_status(self, target_statuses: List[str]) -> bool: instances = self.hook.conn.describe_db_clusters(DBClusterIdentifier=self.id) return instances[0]["Status"] in target_statuses # 3. Use the models in sensor classes (reduce code duplication) class RdsDbSensor(RdsBaseSensor): def _poke(self): return self._resource.check_status(self.target_statuses) @cached_property def _resource(self): if self.db_type == "instance": return RdbInstance(self.hook, self.db_identifier) return RdbCluster(self.hook, self.db_identifier) # 4. Use the models in sensor classes (reduce code duplication) class RdsStartDbOperator(RdsBaseOperator): def _execute(self): self._resource.start() @cached_property def _resource(self): if self.db_type == "instance": return RdbInstance(self.hook, self.db_identifier) return RdbCluster(self.hook, self.db_identifier) ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
