def execute()

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)