def _check_rds_logical_replication()

in src/dma/collector/workflows/readiness_check/_postgres/main.py [0:0]

• 20 lines of code
• 4 McCabe index (conditional complexity)

    def _check_rds_logical_replication(self) -> None:
        rule_code = "RDS_LOGICAL_REPLICATION"
        is_rds = self._is_rds()
        rds_logical_replication = self._get_rds_logical_replication()
        if is_rds:
            for c in self.rule_config:
                if rds_logical_replication != "on":
                    self.save_rule_result(
                        c.db_variant,
                        rule_code,
                        ACTION_REQUIRED,
                        f'`rds.logical_replication` should be set to "on" instead of ({rds_logical_replication})',
                    )
                else:
                    self.save_rule_result(
                        c.db_variant,
                        rule_code,
                        PASS,
                        '`rds.logical_replication` was correctly set to "on"',
                    )