APPLYDJINN
← ETL and Data Engineering: reliable pipelinesProgress & review ↗

Typorithm · medium

Schema versions and quarantine

pythonetldata-engineeringindependent-practice
Saved on this device
Full lesson · 4 learning steps Open any exercise for focused practice
1

Step 1 of 4 · intro card

Schema versions and quarantine

0%

Concept and contract

Normalize known schema versions and retain precise rejection reasons for unknown or invalid records.

A schema contract states which records a pipeline can interpret. The baseline accepts version one, a nonempty string identity and exact integer cents. An omitted version means version one in this deliberately documented source protocol; an unknown explicit version is quarantined rather than guessed.

The output separates canonical accepted records from indexed rejection reasons. Source position makes a bad row traceable without silently mixing it into valid output. This fixture permits signed integer cents: whether a negative amount is meaningful is a separate business rule, not an accidental consequence of parsing JSON.

Mechanism and worked example

The program checks schema version before interpreting its fields, then validates identity and amount.

The program checks schema version before interpreting its fields, then validates identity and amount. Accepted output drops source-only representation details and retains one canonical id and amount_cents shape. Each rejection names the failing rule and the source index so the delivery can be reconciled.

Read the boolean amount case carefully. In Python, bool is a subclass of int, but true is not an acceptable number of cents under this contract. Exact type comparison enforces that boundary. A record that fails an earlier version check is not subsequently reinterpreted using a familiar field name.

Complete worked example

python

def inspect_example(rows):
    accepted, rejected = [], []
    for index, row in enumerate(rows):
        if not isinstance(row, dict):
            rejected.append([index, 'object required'])
            continue
        version = row.get('schema_version', 1)
        if version != 1:
            rejected.append([index, 'unsupported schema'])
            continue
        identity, amount = row.get('id'), row.get('amount_cents')
        if not isinstance(identity, str) or not identity.strip():
            rejected.append([index, 'id required'])
        elif type(amount) is not int:
            rejected.append([index, 'integer cents required'])
        else:
            accepted.append({'id': identity.strip(), 'amount_cents': amount})
    return {'accepted': accepted, 'rejected': rejected}
Reasoning
The program checks schema version before interpreting its fields, then validates identity and amount. Accepted output drops source-only representation details and retains one canonical id and amount_cents shape. Each rejection names the failing rule and the source index so the delivery can be reconciled.
Download the ETL and Data Engineering computer lab ↗

Focused decision: use the complete example for context

python

elif type(amount) is not int:
    rejected.append([index, 'integer cents required'])
Decision
Accept a declared version-two cents field while preserving canonical output, exact integer validation and rejection of unknown schemas.
Download the ETL and Data Engineering computer lab ↗

Counterexamples and common mistakes

The controlled bug replaces exact integer checking with isinstance.

The controlled bug replaces exact integer checking with isinstance. That admits a boolean amount because of Python's numeric inheritance relationship. The resulting record may look plausible in a dictionary while violating the source agreement and contaminating downstream totals.

A second dangerous shortcut would accept every schema version as version one. Field names can change meaning across versions, so similarity is insufficient. Retain the raw object, identify the unsupported version, and require a reviewed normalization rule. Quarantine is a recoverable state with an owner and reconciliation counts, not permission to lose records indefinitely.

Controlled bug: observe the rejected data contract

python

def inspect_example(rows):
    accepted, rejected = [], []
    for index, row in enumerate(rows):
        if not isinstance(row, dict):
            rejected.append([index, 'object required'])
            continue
        version = row.get('schema_version', 1)
        if version != 1:
            rejected.append([index, 'unsupported schema'])
            continue
        identity, amount = row.get('id'), row.get('amount_cents')
        if not isinstance(identity, str) or not identity.strip():
            rejected.append([index, 'id required'])
        elif not isinstance(amount, int):
            rejected.append([index, 'integer cents required'])
        else:
            accepted.append({'id': identity.strip(), 'amount_cents': amount})
    return {'accepted': accepted, 'rejected': rejected}
Reasoning
The controlled bug replaces exact integer checking with isinstance. That admits a boolean amount because of Python's numeric inheritance relationship. The resulting record may look plausible in a dictionary while violating the source agreement and contaminating downstream totals.
Download the ETL and Data Engineering computer lab ↗

Independent practice

Add version two explicitly.

Add version two explicitly. It carries cents in a field named cents and should normalize to the same amount_cents output as version one. Continue rejecting unknown versions, booleans, invalid identities and other incompatible values. Preserve the existing indexed rejection format so operational reports stay comparable.

First widen the allowlist to the two supported versions. Then select the source amount field according to the validated version before running the common exact-type rule. The version-two acceptance case and version-three rejection case distinguish a deliberate migration from simply accepting any new source payload that happens to contain a number.

Step 1: Accept a declared version-two cents field while preserving canonical output, exact integer validation and rejection of unknown schemas. Step 2: Edit the owned exercise file and run verify.py --practice etl-contracts. Preserve caller inputs and the independently declared failure cases.

Practice starter: solve before reading the solution

python

def inspect_example(rows):
    accepted, rejected = [], []
    for index, row in enumerate(rows):
        if not isinstance(row, dict):
            rejected.append([index, 'object required'])
            continue
        version = row.get('schema_version', 1)
        if version != 1:
            rejected.append([index, 'unsupported schema'])
            continue
        identity, amount = row.get('id'), row.get('amount_cents')
        if not isinstance(identity, str) or not identity.strip():
            rejected.append([index, 'id required'])
        elif type(amount) is not int:
            rejected.append([index, 'integer cents required'])
        else:
            accepted.append({'id': identity.strip(), 'amount_cents': amount})
    return {'accepted': accepted, 'rejected': rejected}

# Independent task: Accept a declared version-two cents field while preserving canonical output, exact integer validation and rejection of unknown schemas.
Reasoning
Accept a declared version-two cents field while preserving canonical output, exact integer validation and rejection of unknown schemas.
Download the ETL and Data Engineering computer lab ↗
  • Hint 1: Add version two explicitly. It carries cents in a field named cents and should normalize to the same amount_cents output as version one. Continue rejecting unknown versions, booleans, invalid identities and other incompatible values. Preserve the existing indexed rejection format so operational reports stay comparable.
  • Hint 2: First widen the allowlist to the two supported versions. Then select the source amount field according to the validated version before running the common exact-type rule. The version-two acceptance case and version-three rejection case distinguish a deliberate migration from simply accepting any new source payload that happens to contain a number.

Solution, tests and trade-offs

The solution implements a small versioned adapter boundary: external representations differ, while canonical records stay stable.

The solution implements a small versioned adapter boundary: external representations differ, while canonical records stay stable. This helps downstream deduplication and storage avoid branching on every historical payload format. Independent tests cover both the newly supported version and the rejection behavior that must remain intact.

For larger sources, version-specific validators can be separate functions with an explicit migration table. Decide whether additional fields are preserved, ignored or rejected. Required fields, nullability and numeric ranges should be stated independently; a schema version label cannot make incompatible data valid by itself or substitute for operational monitoring.

Practice solution and acceptance checks

python

def inspect_example(rows):
    accepted, rejected = [], []
    for index, row in enumerate(rows):
        if not isinstance(row, dict):
            rejected.append([index, 'object required'])
            continue
        version = row.get('schema_version', 1)
        if version not in [1, 2]:
            rejected.append([index, 'unsupported schema'])
            continue
        identity, amount = row.get('id'), row.get('amount_cents' if version == 1 else 'cents')
        if not isinstance(identity, str) or not identity.strip():
            rejected.append([index, 'id required'])
        elif type(amount) is not int:
            rejected.append([index, 'integer cents required'])
        else:
            accepted.append({'id': identity.strip(), 'amount_cents': amount})
    return {'accepted': accepted, 'rejected': rejected}
Reasoning
The solution implements a small versioned adapter boundary: external representations differ, while canonical records stay stable. This helps downstream deduplication and storage avoid branching on every historical payload format. Independent tests cover both the newly supported version and the rejection behavior that must remain intact.
Download the ETL and Data Engineering computer lab ↗

Interview reasoning and next step

Describe how you would roll out a source schema change while keeping old deliveries replayable.

Describe how you would roll out a source schema change while keeping old deliveries replayable. Identify the canonical output, supported versions, quarantined counts and the evidence required before backfilling older raw objects. Replaying should use a deliberately chosen adapter version rather than whichever code happens to be newest.

As a stretch, attach the raw object hash and record index to each quarantine entry, then test that accepted plus rejected counts equal the input count. Keep sensitive raw fields out of casual logs. The learning app stores only the authored examples and their prepared results, not external source deliveries.

Optional stretch — after the core task works: Reconcile accepted and rejected counts while retaining raw-object hash and record index.

Computer practice: https://applydjinn.com/typorithm/etl-data-engineering-lab-20261107.zip. Use an isolated Python 3.13 environment, install requirements.txt and run verify.py. Edit exercises/etl-contracts.py and run --practice etl-contracts. Each case owns its files and database; Parquet uses the pinned DuckDB engine. The recovery probe terminates a separate interpreter and reopens the SQLite file. The app retains static examples, prepared results and questions without an editor or code execution. Use the course Download button once to save every lesson and illustration; opening an unsaved course does not automatically download it.

Normalize known schema versions and retain precise rejection reasons for unknown or invalid records.