def _check_tables_without_pk()

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

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

    def _check_tables_without_pk(self, db_name: str, db_check_results: dict[str, dict[str, list]]) -> None:
        rule_code = TABLES_WITH_NO_PK
        result = self.local_db.sql(
            "select CONCAT(nspname, '.', relname) from collection_postgres_tables_with_no_primary_key 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][WARNING].append(f"In database {db_name}, {tables} don't have primary keys")