One Checker -
for item in request.data: value = item.get(request.key_field) if value in seen: duplicates.append(value) else: seen.add(value)
def __init__(self, name: str = "SystemOneChecker"): self.name = name self.checks: List[Callable] = [] self.results: List[OneCheckReport] = []
@app.post("/check-unique", response_model=CheckResponse) async def check_uniqueness(request: CheckRequest): """API endpoint to check if a field is unique across all items""" seen = set() duplicates = [] one checker
def check_single_active_configuration(self, configs: Dict[str, Any]) -> OneCheckReport: """Verify only one configuration is active at a time""" active_configs = [k for k, v in configs.items() if v.get('active', False)] if len(active_configs) == 0: return OneCheckReport( check_name="Single active configuration", result=CheckerResult.ERROR, message="No active configuration found", details={'available_configs': list(configs.keys())} ) elif len(active_configs) == 1: return OneCheckReport( check_name="Single active configuration", result=CheckerResult.PASS, message=f"Exactly one active configuration: {active_configs[0]}" ) else: return OneCheckReport( check_name="Single active configuration", result=CheckerResult.FAIL, message=f"Multiple active configurations found: {active_configs}", details={'active_configs': active_configs} )
# Example 2: Check active configuration configs = { 'dev': {'active': False, 'env': 'development'}, 'prod': {'active': True, 'env': 'production'}, 'staging': {'active': False, 'env': 'staging'} } for item in request
# Run all checks and generate report checker.run_all_checks() print(checker.generate_report())
checker.register_check("Single source of truth", check_sot) v in configs.items() if v.get('active'
return CheckResponse( is_unique=len(duplicates) == 0, duplicates=duplicates, total_items=len(request.data), unique_count=len(seen) )