def _check_tables_replica_identity()

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

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

    def _check_tables_replica_identity(self, db_name: str, db_check_results: dict[str, dict[str, list]]) -> None:
        rule_code = UNSUPPORTED_TABLES_WITH_REPLICA_IDENTITY
        result = self.local_db.sql(
            "select CONCAT(nspname, '.', relname) from collection_postgres_tables_with_primary_key_replica_identity where database_name = $db_name",
            params={"db_name": db_name},
        ).fetchall()
        tables = ", ".join(row[0] for row in result)
        init_results_dict(db_check_results, rule_code)
        if tables:
            db_check_results[rule_code][ACTION_REQUIRED].append(f"{tables} in database {db_name}")