def _check_fdw()

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

• 23 lines of code
• 6 McCabe index (conditional complexity)

    def _check_fdw(self) -> None:
        rule_code = "FDWS"
        result = self.local_db.sql(
            "select foreign_data_wrapper_name as fdw_name, count(distinct table_schema || table_name) as table_count from collection_postgres_table_details where foreign_data_wrapper_name is not null group by foreign_data_wrapper_name"
        ).fetchall()
        fdws = {row[0] for row in result}
        fdw_table_count = {int(row[1]) for row in result}
        for c in self.rule_config:
            unsupported_fdws = fdws.difference(c.supported_fdws)
            for unsupported_fdw, table_count in zip(unsupported_fdws, fdw_table_count):
                self.save_rule_result(
                    c.db_variant,
                    rule_code,
                    ACTION_REQUIRED,
                    f'Unsupported FDW: detected {table_count} "{unsupported_fdw}" foreign tables.',
                )
            if len(unsupported_fdws) == 0:
                self.save_rule_result(
                    c.db_variant,
                    rule_code,
                    PASS,
                    "All utilized foreign data wrappers are supported.",
                )