lesson 227 Reliable Data Pipelines · advanced

Check records against a schema contract

A line can be valid JSON and still be wrong: a delay sent as text, a missing airport, a flag that should be true or false. A schema contract says what every field must be, and validation reports every way a record breaks it.

You'll be able to

Write a schema as a dictionary from field name to type (and whether the field may be empty)
Return every problem in a record, not just the first, and never change the record while checking it
Avoid the bool trap: isinstance(True, int) is True

The idea

Code downstream of ingestion makes assumptions: departure_delay is a number, origin is there, cancelled is true or false. When an upstream system changes (a new export sends delays as text), those assumptions break somewhere far away, in a sum or a chart, with an error that says nothing about the data.

A schema contract writes the assumptions down: the fields a record must have and the type of each. Checking every record against it at the door turns a confusing crash later into a clear report now: *flight 12208: departure_delay should be int, got str*. A validator returns a list of problems; an empty list means the record keeps the contract.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · A schema for flight rows

SCHEMA = {
    "flight_id": int,
    "airline_code": str,
    "origin": str,
    "departure_delay": int,
}

def problems(record):
    found = []
    for field, kind in SCHEMA.items():
        if field not in record:
            found.append(f"missing {field}")
        elif not isinstance(record[field], kind):
            found.append(f"{field}: expected {kind.__name__}, got {type(record[field]).__name__}")
    return found

flights = [
    {"flight_id": 12207, "airline_code": "UA", "origin": "LAS", "departure_delay": -4},
    {"flight_id": 12208, "airline_code": "MQ", "origin": "MCI", "departure_delay": "12"},
    {"flight_id": 12209, "airline_code": "AA", "departure_delay": None},
]
for flight in flights:
    print(flight["flight_id"], problems(flight))

Why bother, when the bad value would fail anyway? Because of *where* it fails. Without a check, the text "3" travels on until arithmetic meets it:

example-2.py · An unchecked type fails far from its cause

events = [
    {"repo": "tidewater/smartqueue", "commits": 2},
    {"repo": "lumen-labs/tinykit", "commits": "3"},
]
total = 0
for event in events:
    total += event["commits"]
print(total)

Real contracts need two more rules. Some fields may legitimately be empty: a cancelled flight has no delay, a WatchEvent has no commits. So each field records whether None is allowed. And bool is a subclass of int in Python, so isinstance(True, int) is True: a contract that says *int* must reject True explicitly, or a flag sent in the wrong field passes.

example-3.py · Optional fields and the bool trap

EVENT_SCHEMA = {
    "id": (str, False),
    "type": (str, False),
    "commits": (int, True),
    "public": (bool, False),
}

def has_type(value, kind):
    if kind is int and isinstance(value, bool):
        return False
    return isinstance(value, kind)

def validate(record, schema):
    errors = []
    for field, (kind, nullable) in schema.items():
        value = record.get(field)
        if value is None:
            if not nullable:
                errors.append(f"{field} is required")
        elif not has_type(value, kind):
            errors.append(f"{field} should be {kind.__name__}")
    return errors

events = [
    {"id": "48000000581", "type": "PushEvent", "commits": 2, "public": True},
    {"id": "48000007504", "type": "WatchEvent", "commits": None, "public": True},
    {"id": "48000002968", "type": "ForkEvent", "commits": True, "public": "yes"},
    {"type": "IssuesEvent", "public": False},
]
for event in events:
    print(validate(event, EVENT_SCHEMA))
print(isinstance(True, int))

When to use it: at every boundary where data enters: after parsing a file, after receiving an API response, before writing to a database. Split the stream into valid records and invalid ones with their problems, like the quarantine in L01. Mistakes to avoid: returning as soon as one problem is found (the sender fixes one field, resends, and fails again); fixing values while validating (a validator reports, it doesn't change the record); and forgetting that True is an int.

Review

A schema contract is data: field → type, and whether the field may be empty.
Validate every field and return every problem; never change the record while checking it.
True is an int to isinstance: reject bools explicitly in number fields. CSV values are text, so convert by the schema.
Next: valid records can still be loaded twice. Idempotent loads make that harmless.

Practise this