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.

Easy Python 3.9+ Aug 9, 2026 Data pipelines & processing 13 views 0 copies

Python code

39 lines
Python 3.9+
from 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

stdout
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

  1. Use pydantic's BaseModel for a more declarative schema with automatic validation and error reporting
  2. 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

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Data pipelines & processing

Related tutorials and quizzes for this topic.