Validate dict schema at pipeline boundary in Python
This code validates a dictionary against a TypedDict schema at a pipeline boundary, enforcing required fields and types with custom error messages.
Python code
39 linesfrom typing import Any, TypedDict
class Person(TypedDict):
name: str
age: int
email: str
def validate_person(data: dict[str, Any]) -> Person:
errors: list[str] = []
if not isinstance(data.get("name"), str) or not data["name"].strip():
errors.append("name must be a non-empty string")
if not isinstance(data.get("age"), int) or data["age"] < 0:
errors.append("age must be a non-negative integer")
if not isinstance(data.get("email"), str) or "@" not in data["email"]:
errors.append("email must be a valid string containing '@'")
if errors:
raise ValueError(f"Validation failed: {'; '.join(errors)}")
return {
"name": data["name"].strip(),
"age": data["age"],
"email": data["email"].strip(),
}
if __name__ == "__main__":
valid_input = {"name": " Alice ", "age": 30, "email": "alice@example.com"}
invalid_input = {"name": "", "age": -5, "email": "not-an-email"}
print("Valid input:", validate_person(valid_input))
try:
validate_person(invalid_input)
except ValueError as exc:
print("Invalid input error:", exc)
Output
Valid input: {'name': 'Alice', 'age': 30, 'email': 'alice@example.com'}
Invalid input error: Validation failed: name must be a non-empty string; age must be a non-negative integer; email must be a valid string containing '@'
How it works
The validate_person function checks each required field with isinstance to enforce types and additional constraints like non-empty strings and valid email format. It accumulates all errors instead of failing on the first one, giving callers a complete picture of what went wrong. The function returns a new dictionary with stripped values, ensuring downstream code receives clean data. This pattern is useful at pipeline boundaries where data from external sources needs to be normalized and validated before further processing.
Common mistakes
- Forgetting to check for missing keys before accessing them in the condition, causing KeyError
- Using `type(data['age']) is int` instead of `isinstance`, which fails for bool subclasses
- Not stripping values after validation, passing through extra whitespace
Variations
- Use pydantic's BaseModel for a more declarative schema with automatic validation and error reporting
- Implement a generic validation function that takes a schema dict and iterates over fields
Real-world use cases
- Sanitizing and validating API request payloads before inserting into a database.
- Ensuring data files from partner systems meet the required format before processing in an ETL job.
- Verifying configuration dictionaries loaded from JSON files against a defined schema at service startup.
Sponsored
More from Data pipelines & processing
- Add a UUID Surrogate Key to Each Row in a CSV with Python easy
- Attach Source File Metadata to Records in Python easy
- Build a Python Utility That Detects Duplicate Records Across Multiple Excel Sheets medium
- Check Null Rate Threshold in PySpark DataFrame medium
- Count Records Processed per Category in Python easy
- Create Data Helper Functions in Python for Beginners easy
Keep learning
Related tutorials and quizzes for this topic.