checker.register_check("User IDs uniqueness", check_user_ids)
class CheckerResult(Enum): """Possible results from a one-check operation""" PASS = "pass" FAIL = "fail" WARNING = "warning" ERROR = "error"
def check_one_to_one_relationship(self, left_data: List, right_data: List, left_key: str, right_key: str) -> OneCheckReport: """Check that relationships are one-to-one, not one-to-many""" left_keys = [item[left_key] for item in left_data] right_keys = [item[right_key] for item in right_data] # Check if any left key maps to multiple right items from collections import Counter right_counts = Counter(right_keys) violations = [] for key in set(left_keys): if right_counts.get(key, 0) > 1: violations.append({ 'key': key, 'count': right_counts[key] }) if violations: return OneCheckReport( check_name="One-to-one relationship check", result=CheckerResult.FAIL, message=f"Found {len(violations)} keys with multiple relationships", details={'violations': violations} ) else: return OneCheckReport( check_name="One-to-one relationship check", result=CheckerResult.PASS, message="All relationships are one-to-one" ) one checker
def check_sot(): return checker.check_single_source_of_truth(sources)
# Example 3: Check one-to-one relationship users = [{'user_id': 1, 'name': 'Alice'}, {'user_id': 2, 'name': 'Bob'}] profiles = [{'user_id': 1, 'bio': 'Engineer'}, {'user_id': 1, 'bio': 'Also artist'}] # Multiple profiles for user 1 checker
def check_relationships(): return checker.check_one_to_one_relationship(users, profiles, 'user_id', 'user_id')
def generate_report(self) -> str: """Generate a formatted report of all check results""" if not self.results: return "No checks have been run yet." summary = { CheckerResult.PASS: 0, CheckerResult.FAIL: 0, CheckerResult.WARNING: 0, CheckerResult.ERROR: 0 } report_lines = [f"\n{'='*60}", f"One Checker Report: {self.name}", f"{'='*60}\n"] for result in self.results: summary[result.result] += 1 status_icon = { CheckerResult.PASS: "✓", CheckerResult.FAIL: "✗", CheckerResult.WARNING: "⚠", CheckerResult.ERROR: "!" }.get(result.result, "?") report_lines.append(f"{status_icon} [{result.result.value.upper()}] {result.check_name}") report_lines.append(f" {result.message}") if result.details: report_lines.append(f" Details: {json.dumps(result.details, indent=2)}") report_lines.append("") report_lines.append(f"{'='*60}") report_lines.append(f"Summary: {summary[CheckerResult.PASS]} passed, " f"{summary[CheckerResult.FAIL]} failed, " f"{summary[CheckerResult.WARNING]} warnings, " f"{summary[CheckerResult.ERROR]} errors") return "\n".join(report_lines) if name == " main ": # Create a OneChecker instance checker = OneChecker("DataConsistencyChecker") checker.register_check("User IDs uniqueness"
checker.register_check("Single source of truth", check_sot)