in src/dma/collector/workflows/readiness_check/_postgres/main.py [0:0]
def execute(self) -> None:
"""Execute postgres checks"""
self._check_version()
self._check_collation()
self._check_rds_logical_replication()
self._check_wal_level()
self._check_max_replication_slots()
self._check_max_wal_senders_replication_slots()
self._check_max_worker_processes()
self._check_fdw()
# Per DB Checks.
self._check_extensions()
self._check_replication_role()
# db_check_results stores the verification results for all DBs.
for config in self.rule_config:
db_check_results: dict[str, dict[str, list]] = {}
for db in sorted(self.get_all_dbs()):
is_pglogical_installed = self._check_pglogical_installed(db, db_check_results)
if is_pglogical_installed:
privilege_check_passed = self._check_privileges(db, db_check_results)
if not privilege_check_passed:
continue
self._check_if_node_exists(db, db_check_results)
self._check_tables_without_pk(db, db_check_results)
self._check_tables_replica_identity(db, db_check_results)
self._save_results(config.db_variant, db_check_results)