When working with external APIs or unpredictable data sources, you often need to enforce structure on responses that might be inconsistent or malformed. This is especially important when dealing with language models, which can produce varied output formats even with specific instructions.
Schema validation is the process of checking whether data conforms to a predefined structure. A schema defines what fields are required, what types they should be, and any constraints they must meet. Think of it as a contract that your data must fulfill.
Retry patterns are essential when dealing with unreliable operations. Instead of failing immediately on the first error, you give the operation multiple chances to succeed. This is particularly useful for network requests, external API calls, or any operation that might temporarily fail but could succeed on subsequent attempts.
The combination of these patterns creates robust systems that can handle unpredictable inputs while maintaining data integrity. When an operation fails validation, you retry it rather than immediately giving up. This gives you the best of both worlds: strict data requirements and resilience to temporary failures.
A common implementation strategy is to separate concerns: one function handles the unreliable operation (like calling an API), another validates the result, and a third orchestrates the retry logic. This separation makes each component easier to test and maintain.
The key insight is that validation should happen after each attempt, not just at the end. This allows you to make informed decisions about whether to retry (invalid format) or fail fast (impossible request).
1import json
2
3def validate_user_data(data, required_fields):
4 if not isinstance(data, dict):
5 return False
6
7 for field in required_fields:
8 if field not in data:
9 return False
10 return True
11
12def fetch_with_retry(url, validator, max_attempts=3):
13 for attempt in range(max_attempts):
14 try:
15 # Simulated API call
16 response = make_api_call(url)
17 parsed_data = json.loads(response)
18
19 if validator(parsed_data):
20 return parsed_data
21
22 except (json.JSONDecodeError, ConnectionError) as e:
23 if attempt == max_attempts - 1:
24 raise Exception(f"Failed after {max_attempts} attempts")
25 continue
26
27 raise Exception("All validation attempts failed")