def _check_max_wal_senders_replication_slots()

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

• 27 lines of code
• 5 McCabe index (conditional complexity)

    def _check_max_wal_senders_replication_slots(self) -> None:
        rule_code = "WAL_SENDERS_REPLICATION_SLOTS"
        wal_senders_result = self.local_db.sql(
            "select c.setting_value as max_wal_senders from collection_postgres_settings c where c.setting_name='max_wal_senders';"
        ).fetchone()
        wal_senders = int(wal_senders_result[0]) if wal_senders_result is not None else 0
        total_replication_slots_result = self.local_db.sql(
            "select c.setting_value as max_replication_slots from collection_postgres_settings c where c.setting_name='max_replication_slots';"
        ).fetchone()
        total_replication_slots = (
            int(total_replication_slots_result[0]) if total_replication_slots_result is not None else 0
        )
        for c in self.rule_config:
            if wal_senders < total_replication_slots:
                self.save_rule_result(
                    c.db_variant,
                    rule_code,
                    ACTION_REQUIRED,
                    f"Insufficient `max_wal_senders`: {wal_senders}, should be set to at least the same as `max_replication_slots`: {total_replication_slots}.",
                )
            else:
                self.save_rule_result(
                    c.db_variant,
                    rule_code,
                    PASS,
                    f"`max_wal_senders` current value: {wal_senders}, this meets or exceeds the `max_replication_slots` value of {total_replication_slots}",
                )