def _check_pglogical_installed()

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


    def _check_pglogical_installed(self, db_name: str, db_check_results: dict[str, dict[str, list]]) -> bool:
        rule_code = PGLOGICAL_INSTALLED
        result = self.local_db.sql(
            query="select count(*) from collection_postgres_extensions where extension_name = 'pglogical' and database_name = $db_name",
            params={"db_name": db_name},
        ).fetchone()
        is_installed = result[0] > 0 if result is not None else False
        init_results_dict(db_check_results, rule_code)
        if not is_installed:
            db_check_results[rule_code][ACTION_REQUIRED].append(db_name)
        else:
            db_check_results[rule_code][PASS].append(db_name)
        return is_installed