def _check_if_node_exists()

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

• 12 lines of code
• 3 McCabe index (conditional complexity)

    def _check_if_node_exists(self, db_name: str, db_check_results: dict[str, dict[str, list]]) -> None:
        rule_code = PGLOGICAL_NODE_ALREADY_EXISTS
        result = self.local_db.sql(
            "select count(*) from collection_postgres_pglogical_provider_node where database_name = $db_name",
            params={"db_name": db_name},
        ).fetchone()
        node_exists = result[0] > 0 if result is not None else False
        init_results_dict(db_check_results, rule_code)
        if node_exists:
            db_check_results[rule_code][ACTION_REQUIRED].append(db_name)
        else:
            db_check_results[rule_code][PASS].append(db_name)