def _check_privileges()

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

• 13 lines of code
• 2 McCabe index (conditional complexity)

    def _check_privileges(self, db_name: str, db_check_results: dict[str, dict[str, list]]) -> bool:
        rule_code = PRIVILEGES
        errors = self._check_pglogical_privileges(db_name)

        errors.extend(self.check_user_obj_privileges(db_name))
        all_errors = "\n".join(errors)
        init_results_dict(db_check_results, rule_code)
        if len(errors) > 0:
            db_check_results[rule_code][ACTION_REQUIRED].append(f"{all_errors} in database {db_name}")
        else:
            db_check_results[rule_code][PASS].append(
                f"User has all privileges required for migration for the database {db_name}"
            )
        return len(errors) == 0