Python Fundamentals for Data Engineering

Data Engineering & Modern Data Platforms · Lesson 04 · Data Engineering Foundations

Lesson Overview

The contract checker in lesson 1 failed in production for a reason that a unit test on a laptop did not catch. A batch file contained one progress event whose attempt_number arrived as the JSON number 1.0, and another whose timestamp used a space instead of a T. The checker coerced both, wrote them, and moved on. A third event contained the string "1" for the attempt. That one was rejected. The raw zone now held a mix of strict and coerced rows, and the completion job's notion of attempt 1 depended on which file the event had been in.

The defect was not "Python is loose." The defect was a pipeline that used dictionaries, truthiness, and implicit conversions as a data contract. Data engineering Python is ordinary Python aimed at a hostile input: files you did not write, types you must not guess, and failures you must count. This lesson is that discipline. It is not a second copy of /learn/python. The primer owns syntax for its own sake. Here every construct has to earn its place on the path from a newline-delimited JSON file to a batch the raw zone is allowed to keep.

You will use Python 3.11 or newer and the standard library. No pandas, no Spark, no warehouse client. Those tools are justified later, and they hide the grain if you reach for them while the contract is still soft. Lesson 5 adds generators, context managers, concurrency, packaging, logging, and tests as a suite. This lesson writes the types, the collections, the functions, and the exceptions those tools will wrap. Lesson 6 is where the accepted batch becomes SQL. If you cannot explain a row before it hits the database, the database will explain it for you, in an incident.

Planning assumptions stay labeled. A teaching batch fits in memory. A peak day of about fifty million events does not, which is why lesson 5 refuses to hold the whole file in a list. The functions here are written so they can later accept an iterator without changing their rules.

Learning Objectives

By the end of this lesson you will be able to:

1. Choose a Python type for a progress-event field and explain what a wrong choice does to a raw zone.
2. Use lists, tuples, dictionaries, and sets for the jobs they are actually good at in a batch pipeline.
3. Write a pure function that parses one record and a second function that reviews a batch without mutating the caller's collection.
4. Separate contract failures from programmer errors with a small exception hierarchy.
5. Apply type hints that a reviewer can check, including Mapping, union types, and a frozen dataclass.
6. Reject naive datetimes, boolean-as-integer traps, and mutable default arguments in pipeline code.
7. Split the contract, the file reader, and the summary into modules with a clear import direction.
8. Decide when Python is the right tool for a step and when the step belongs in SQL.

Prerequisites

• Lessons 1 through 3: the progress-event contract, the raw-zone boundary, and the data engineer as the owner of that boundary. This lesson implements the boundary. It does not reopen who is accountable.
• You can read Python functions and if statements. If decorators, generators, or packaging are unfamiliar, leave them for lesson 5 and the primer at /learn/python.
• You can run python3 --version and expect 3.11 or newer. The examples use list[str], str | None, and datetime with zones. They do not use third-party type checkers, though a checker is recommended in the practices section.
• You do not need a Jupyter notebook. Pipelines that ship are modules you can import and test. A notebook is a fine scratch pad and a poor system of record for a contract.

Detailed Core Concepts

Why Python is on this path

Python is the language of the extract, the contract, the file layout, and the orchestration glue in this course because the standard library can read JSON, talk to files, and express a rule without a cluster. It is a poor language for the official aggregation once the grain is relational and the consumer is SQL. Both statements can be true. The data engineer who implements completion in a Python for loop and the data engineer who loads untyped JSON straight into a warehouse are making the same mistake at different layers: they are putting the contract in the place where it is hardest to test.

Use Python at the boundary when:

• The input is files, HTTP pages, or a library the warehouse cannot call.
• The rule is about one record's shape, or about a batch's uniqueness, and you want a unit test that does not boot a database.
• The failure must be a typed result the job can count.

Use SQL, in lesson 6, when:

• The rule is a join, a filter, or an aggregate over a grain the database already stores.
• The consumer of the result is a person with a SQL tool.
• You would otherwise reimplement GROUP BY and get it wrong.

Use neither to coerce types silently. int("1") is a decision. int(1.0) is also a decision, and it hides a producer that emitted a float. The contract in this lesson accepts JSON integers only.

Values, types, and the contract

A Python value has a type at runtime whether or not you annotate it. Annotations are for humans and for checkers. The raw zone cares about runtime behavior.

The boolean trap is the one that survives code review, because a check written as isinstance(value, int) accepts True, and True == 1 in Python. Attempt number one and attempt number "yes" become the same key. The parser in this lesson uses type(value) is int. That is intentionally strict. It is not a style preference. It is the grain.

None means absence. It is not zero, not the empty string, and not the string "null". A missing checkpoint on a completed event may be legitimate. A missing lesson_id is not. Model those differently: an optional field is int | None after a successful parse, and a required field that arrives as JSON null is a ContractError. Callers do not use or to default a missing lesson id to a placeholder. payload.get("lesson_id") or "" turns both null and "" into "", and it also turns a hypothetical valid id of 0 into "" if you ever used it on a number. Defaults hide the producer bug the quarantine exists to count.

Collections and what they promise

List. Ordered, mutable, allows duplicates. Use a list for the sequence of violations, where order is the row order a human will read, and duplicates are two failures. Do not use a list as the uniqueness index for event_id. Membership tests on a list are linear, and a peak-day file will make that obvious too late.

Tuple. Ordered, immutable. Use a tuple for a key (learner_id, lesson_id, attempt_number) once the record is accepted, and for the return value of a review when you want the caller to stop appending. A tuple is not a record with names. Passing a five-element tuple through three functions is how columns swap places. Prefer a frozen dataclass once the record has more than a couple of fields.

Dict. Keyed lookup. Use a dict to index accepted events by event_id inside a batch. Remember that dict keys must be hashable and that the dict remembers insertion order. It does not validate that two logical events are the same. You write that comparison. A defaultdict(list) is useful for grouping violations by code. It is harmful as a function default argument, covered below.

Set. Unordered uniqueness. Use a set for "prohibited keys we have seen" or for a membership test of allowed event types, built once as a frozenset so nobody adds debug at runtime. Do not use a set as the only store of events if you still need the first payload to compare against a conflicting duplicate. A set of ids tells you "seen." It does not tell you "seen with a different body."

JSON objects become Python dict instances via the standard library json module. They are dict[str, object] until you parse them. Treat them as untrusted mappings. Do not subclass them. Do not pass the raw dict into the transform that computes completion.

Functions, purity, and mutation

A pure function returns a value and does not change its arguments or any global store. parse_event should be pure: same mapping in, same event or the same exception out, no write to a file. Purity is what makes a contract test fast and deterministic. The function that reads a path is not pure, and it should do nothing except read bytes and call the pure function.

Mutation is the usual way a batch review lies. If review_batch appends to a list the caller also holds, a retry of the review sees the previous run's rows. Take an iterable, build new collections, return them. The lesson 1 checker already returned tuples. Keep that habit.

Default arguments are evaluated once, at function definition. This function does not do what the signature suggests:

pythonCopy
def add_violation(item, sink=[]):  # the same list for every call
    sink.append(item)
    return sink

The second call still contains the first item. In a long-running worker, that list is a cross-batch leak and a privacy defect if the items hold learner ids. Use sink: list | None = None and create the list inside the function. The same rule applies to dicts and sets. Lesson 5's worker will run for hours. A mutable default will survive every file it processes.

Exceptions are part of the contract

Python will raise KeyError, ValueError, and TypeError for free. A pipeline that lets those escape has lost the row number and the field. A pipeline that catches Exception has also swallowed MemoryError and KeyboardInterrupt's relatives poorly, and it will mark a programming bug as a bad record.

Use a small hierarchy:

• ContractError for a record the producer got wrong. The batch review catches this and continues, then rejects the batch if any exist.
• Let KeyError and TypeError that indicate a bug in the parser escape during development. Inside parse_event, translate expected shape problems into ContractError so the caller has one type to catch.
• Do not catch Exception in the per-row loop.

ContractError carries a code (MISSING_FIELD, BAD_TYPE, PROHIBITED_FIELD, BAD_TIME) so metrics can count without parsing English. The message can include the field name. It must not include a raw email value if the failure is that the email was present. Say the key name. Do not echo the value. The lesson 1 checker was slightly careless about messages. This lesson tightens that.

A batch policy remains: any ContractError rejects the batch. Identical duplicate event_id values are not errors. Conflicting duplicates are ContractError with code ID_CONFLICT. The policy is a product decision from lesson 1. Python's job is to make the policy obvious in the control flow, not to "be helpful" by skipping bad rows.

Modules and import direction

A single file is fine for a lesson and wrong for a job that three roles will edit. Split by who changes the file.

Import direction points inward, toward the contract. ndjson.py may import batch.py. contract.py may not import ndjson.py. A cycle means a rule change risks a behavior change in file framing, and reviewers cannot see the blast radius. The data architect hat from lesson 3 cares about this for the same reason they care about writers to the mart: one direction, one owner.

if __name__ == "__main__" belongs on a thin command module, not on the contract. Importing a module should not parse a file and should not print. Tests import the contract. A print on import pollutes the test output and, worse, runs work the test did not ask for.

Type hints a reviewer can believe

Hints do not execute in CPython unless you force them to. event_id: str will not stop a caller from passing an int. They still belong on every public function in a pipeline, because the reviewer is checking the contract by eye and because a type checker will catch the boolean mistake's cousins at commit time.

Prefer:

• Mapping[str, object] for a JSON object you have not parsed. Mapping signals that you will not mutate it. dict signals that you might.
• datetime only after it is known to be aware. You cannot say "aware datetime" in the standard type system. Encode it in the constructor and do not accept datetime from outside without a check.
• tuple[ProgressEvent, ...] for an immutable sequence of unknown length. list if the caller is expected to build it.
• str | None for an optional identifier on a violation, when the row had no parseable id. Do not use a string sentinel "unknown" that can collide with a real id.

Avoid Any. It switches the hints off at the exact boundary where the input is hostile. object, followed by isinstance checks, is honest. Any is a shrug.

This course does not require a particular checker. pyright and mypy both understand the annotations used here. Run one of them in CI when the repository from lesson 5 exists. A green test suite without a checker will still accept a function annotated as returning ProgressEvent that sometimes returns None.

Dataclasses, equality, and identity

@dataclass(frozen=True) gives you a constructor, an equality method, and a hash, and it refuses attribute assignment. Frozen events can be dict keys. Mutable events cannot, or they can and then corrupt the dict if you change a field. Progress events are facts once parsed. Freeze them.

Equality is field equality. Two ProgressEvent values with the same fields compare equal, which is what the duplicate check wants: same id and previous != event means a conflict. Do not implement equality on event_id alone. That would make a conflict compare equal and disappear.

event_id is the identity the producer claimed. The dataclass equality is the payload identity. The batch logic uses both. Mixing them into one __eq__ method saves a line and loses the distinction the quarantine is built on.

Time

datetime.fromisoformat accepts several ISO-8601 forms. It does not accept the Z suffix on versions before 3.11. This course requires 3.11 or newer, where Z is accepted as UTC. Do not replace("Z", "+00:00") unless you are also supporting an older interpreter you have named. Do not strip the offset and call the result UTC. A timestamp of 2026-03-02T14:02:11+05:30 is a real instant and must be converted with astimezone(timezone.utc), not truncated.

Naive datetimes compare in a way that ignores the missing zone and raise when compared to aware ones in some operations and not in others, depending on the call. Ban them at the constructor. Store UTC. The business zone for "Tuesday" is a property of the metric spec, applied in SQL, not a property of the file.

When the standard library is the design

json parses one value. Newline-delimited JSON is a framing you write: one JSON object per line, blank lines forbidden or explicitly skipped, a line that is two objects concatenated is a contract failure. Do not json.load an entire array if the producer promised NDJSON. A single array cannot be rejected per line without holding the file, and a trailing comma from a tool that "helpfully" wrote an array will fail the whole day with a message that points at the end.

pathlib.Path is the path object to use. String concatenation of directories is how a batch parameter becomes a path traversal bug. The reader in this lesson takes a path the caller resolved. It does not accept a URL. HTTP ingestion is a later lesson. Mixing it in here would mean catching network errors in the same loop as bad integers, and the severity table from lesson 3 would not know which class it was.

Architecture and Design Diagrams

The data flow inside the process is short on purpose. Bytes become a mapping, the mapping becomes an event or a violation, and only an empty violation list may leave the process toward the raw zone.

mermaidCopy
flowchart LR
    file[NDJSON file on local disk]
    lines[Lines as text]
    maps[JSON objects as mappings]
    review[Batch review]
    events[Frozen ProgressEvent values]
    violations[ContractError list]
    raw[(Raw zone write, later)]
    file --> lines --> maps --> review
    review --> events
    review --> violations
    events -->|only if violations is empty| raw
    violations -->|any| stop[Reject batch, write nothing]

Module direction is the other diagram reviewers should see in the repository, because it is the one that prevents the metric from leaking into the parser.

mermaidCopy
flowchart LR
    cli[cli.py]
    ndjson[ndjson.py]
    batch[batch.py]
    contract[contract.py]
    summary[summary.py]
    cli --> ndjson
    cli --> summary
    ndjson --> batch
    batch --> contract
    summary --> batch

There is no arrow from contract.py to a database, and none from summary.py to the player. The completion mart is downstream of a file this process either accepts or refuses. Lesson 6 reads accepted rows. It does not import these modules at query time.

A single bad record's control flow:

mermaidCopy
flowchart TD
    row[One JSON object]
    banned{Prohibited key?}
    types{Required types exact?}
    built[ProgressEvent frozen]
    dup{event_id already seen?}
    same{Payload equal?}
    keep[Keep the first]
    conflict[ID_CONFLICT]
    err[ContractError, remember row index]
    row --> banned
    banned -->|yes| err
    banned -->|no| types
    types -->|no| err
    types -->|yes| built --> dup
    dup -->|no| keep
    dup -->|yes| same
    same -->|yes| keep
    same -->|no| conflict

Technical Explanation

Parsing without coercing

The standard library's json.loads is the right parser. It maps JSON numbers with no fraction to int and numbers with a fraction to float. It maps true to True. It does not map the string "1" to an int. That is the behavior you want, followed by a strict type check. Custom JSONDecoder hooks that "normalize" numbers will reintroduce coercion. Do not install one.

Text encoding is UTF-8. A file with a byte-order mark or a Latin-1 exception is a batch failure of the framing layer, not a row to skip. Open with encoding="utf-8" and do not pass errors="replace", which would turn bad bytes into a learner id that never existed. Replacement characters in an identifier are a data-loss function disguised as resilience.

Line numbers in violation messages are 1-based, because that is what an editor shows. Internal indexes can be 0-based if you convert at the message. Pick one and test it. An off-by-one in a quarantine message sends the on-call engineer to the wrong event, and they will "fix" a good row.

Uniqueness without losing the payload

A dict from event_id to ProgressEvent is the index. When the id is new, store it. When the id exists and the frozen events compare equal, count a duplicate delivery and do not append a second event. When they differ, record ID_CONFLICT and do not keep either as accepted. "Keep the first" on a conflict is a policy you could choose. This course does not. A conflict means the producer used one id for two facts, and choosing the first hides the second from the count of accepted rows while still storing something. Rejecting the batch forces the producer to repair the id. The cost is a delayed day. The cost of the alternative is a raw zone that picked a winner silently.

Memory: the dict holds every unique event in the batch. That is acceptable for the teaching file and for a batch you have already decided fits in memory. Lesson 5 streams events and still needs an id index if the uniqueness rule is "within the file." An index of ids alone, plus a hash of the canonical payload, is the production form when the event body is large. Do not build that optimization here. Do write the rule so the index is local to the function and dies when the function returns.

Strings and identifiers

Identifiers are str after they are known to be strings, not numbers that you format. A JSON number 1042 is not a learner id even if the current ids look numeric. Reject non-strings. Leading and trailing whitespace is a contract failure, not a strip. Stripping hides a producer that pads ids, and "lrn_1042" and "lrn_1042 " would become one learner after the strip and two learners in the operational database. The player does not strip. The checker must not be cleverer than the player.

Email rejection stays structural: a key named email, a key ending in _email, or an @ inside learner_id. Do not try to detect emails in free-text fields you do not have. A detector that guesses will false-positive on a lesson id someday and false-negative on a field named contact. The prohibited-key list is the governance artifact. The code loads it as a frozenset.

Summaries are not metrics

The review returns counts: rows seen, accepted, duplicate deliveries, violations by code. Those counts are operational telemetry for the pipeline owner. They are not the completion rate. Printing "accepted=1000" next to a dashboard tempts someone to divide it by enrollments. The summary object has no method called completion_rate. If you feel the absence, the absence is the design.

Failure and exit codes

A command that rejects a batch exits with status 2. A command that cannot read the file exits with status 1. A programming bug should traceback and exit non-zero without being relabeled as a contract failure. Lesson 3's severity table depends on that distinction. SystemExit is appropriate in cli.py only. Library modules raise. They do not call sys.exit, because a caller in a test or an orchestrator cannot catch a process exit cleanly if you have mixed it into the parser.

What type hints will not save you from

Hints do not check JSON. They do not make json.loads return Mapping[str, object]; it returns Any in the stubs, and you narrow it. They do not stop a colleague from calling dict(event) and mutating a copy they think is the original. Frozen dataclasses stop attribute assignment on the event. They do not deep-freeze a list you stored inside one. Do not store lists inside ProgressEvent. If a future field is a collection, store a tuple.

They also do not replace the prohibited-field test. An annotated ProgressEvent simply has no email attribute. The attack is the extra key on the mapping, which annotations never see. The loop over keys is the control. Annotations document the accepted shape after that loop has passed.

Practical Implementation or Design Walkthrough

Problem

Replace the lesson 1 checker with a stricter parser that reads NDJSON, refuses coercion, and returns a summary. The batch is still accepted or rejected as a whole.

Sample records

Save the following as sample.ndjson. The third line is a conflict on evt_1001. The fourth line is a float attempt. The file must be rejected, and a clean file of the first line twice must be accepted once.

textCopy
{"event_id": "evt_1001", "learner_id": "lrn_1042", "lesson_id": "les_07", "event_type": "completed", "attempt_number": 1, "event_time": "2026-03-02T14:02:11Z"}
{"event_id": "evt_1001", "learner_id": "lrn_1042", "lesson_id": "les_07", "event_type": "completed", "attempt_number": 1, "event_time": "2026-03-02T14:02:11Z"}
{"event_id": "evt_1001", "learner_id": "lrn_1042", "lesson_id": "les_07", "event_type": "checkpoint", "attempt_number": 1, "event_time": "2026-03-02T14:02:11Z"}
{"event_id": "evt_1004", "learner_id": "lrn_1042", "lesson_id": "les_07", "event_type": "checkpoint", "attempt_number": 1.0, "event_time": "2026-03-02T15:00:00+00:00"}

Code

One module is shown so the lesson can be run as a single file. The comments mark the split from the module diagram. Python 3.11+. Standard library only.

pythonCopy
"""Strict NDJSON contract for progress events. Python 3.11+."""

from __future__ import annotations

import json
import sys
from collections import Counter
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Mapping, TextIO

ALLOWED_EVENT_TYPES = frozenset({"checkpoint", "completed"})
PROHIBITED_KEYS = frozenset({"email", "full_name", "card_token"})


class ContractError(Exception):
    def __init__(self, code: str, message: str, event_id: str | None = None) -> None:
        super().__init__(message)
        self.code = code
        self.event_id = event_id


@dataclass(frozen=True)
class ProgressEvent:
    event_id: str
    learner_id: str
    lesson_id: str
    event_type: str
    attempt_number: int
    event_time: datetime

    def __post_init__(self) -> None:
        if type(self.attempt_number) is not int or self.attempt_number < 1:
            raise ContractError("BAD_TYPE", "attempt_number must be an int >= 1", self.event_id)
        if self.event_time.tzinfo is None or self.event_time.utcoffset() is None:
            raise ContractError("BAD_TIME", "event_time must be timezone-aware", self.event_id)


@dataclass(frozen=True)
class BatchSummary:
    seen: int
    accepted: int
    duplicate_deliveries: int
    violations: tuple[ContractError, ...]

    @property
    def ok(self) -> bool:
        return not self.violations


def _require_str(payload: Mapping[str, object], field: str, event_id: str | None) -> str:
    value = payload.get(field)
    if type(value) is not str or value.strip() != value or value == "":
        raise ContractError("BAD_TYPE", f"{field} must be a non-empty string without padding", event_id)
    return value


def parse_event(payload: Mapping[str, object]) -> ProgressEvent:
    for key in payload:
        if key in PROHIBITED_KEYS or key.endswith("_email"):
            raise ContractError("PROHIBITED_FIELD", f"prohibited field present: {key}")
    raw_id = payload.get("event_id")
    event_id = raw_id if type(raw_id) is str else None
    learner_id = _require_str(payload, "learner_id", event_id)
    if "@" in learner_id:
        raise ContractError("PROHIBITED_FIELD", "learner_id must not contain @", event_id)
    lesson_id = _require_str(payload, "lesson_id", event_id)
    event_key = _require_str(payload, "event_id", event_id)
    event_type = _require_str(payload, "event_type", event_key)
    if event_type not in ALLOWED_EVENT_TYPES:
        raise ContractError("BAD_TYPE", "unsupported event_type", event_key)
    attempt = payload.get("attempt_number")
    if type(attempt) is not int or attempt < 1:
        raise ContractError("BAD_TYPE", "attempt_number must be a JSON integer >= 1", event_key)
    raw_time = payload.get("event_time")
    if type(raw_time) is not str:
        raise ContractError("BAD_TIME", "event_time must be a string", event_key)
    try:
        parsed = datetime.fromisoformat(raw_time)
    except ValueError as exc:
        raise ContractError("BAD_TIME", "event_time is not ISO-8601", event_key) from exc
    if parsed.tzinfo is None:
        raise ContractError("BAD_TIME", "event_time must include an offset", event_key)
    return ProgressEvent(
        event_id=event_key,
        learner_id=learner_id,
        lesson_id=lesson_id,
        event_type=event_type,
        attempt_number=attempt,
        event_time=parsed.astimezone(timezone.utc),
    )


def review_mappings(payloads: list[Mapping[str, object]]) -> tuple[tuple[ProgressEvent, ...], BatchSummary]:
    accepted: list[ProgressEvent] = []
    seen: dict[str, ProgressEvent] = {}
    violations: list[ContractError] = []
    duplicates = 0
    for index, payload in enumerate(payloads, start=1):
        try:
            event = parse_event(payload)
        except ContractError as exc:
            violations.append(exc)
            continue
        previous = seen.get(event.event_id)
        if previous is None:
            seen[event.event_id] = event
            accepted.append(event)
        elif previous != event:
            violations.append(ContractError("ID_CONFLICT", f"line {index}: conflicting event_id", event.event_id))
        else:
            duplicates += 1
    summary = BatchSummary(len(payloads), len(accepted) if not violations else 0, duplicates, tuple(violations))
    if violations:
        return tuple(), summary
    return tuple(accepted), summary


def load_ndjson(handle: TextIO) -> list[Mapping[str, object]]:
    payloads: list[Mapping[str, object]] = []
    for line_number, line in enumerate(handle, start=1):
        if line.strip() == "":
            raise ContractError("BAD_FRAMING", f"line {line_number}: blank line")
        try:
            value = json.loads(line)
        except json.JSONDecodeError as exc:
            raise ContractError("BAD_FRAMING", f"line {line_number}: invalid JSON") from exc
        if type(value) is not dict:
            raise ContractError("BAD_FRAMING", f"line {line_number}: expected an object")
        payloads.append(value)
    return payloads


def violation_codes(summary: BatchSummary) -> Counter[str]:
    return Counter(item.code for item in summary.violations)


def _demo() -> None:
    good = {
        "event_id": "evt_1001",
        "learner_id": "lrn_1042",
        "lesson_id": "les_07",
        "event_type": "completed",
        "attempt_number": 1,
        "event_time": "2026-03-02T14:02:11Z",
    }
    events, summary = review_mappings([good, dict(good)])
    assert summary.ok and len(events) == 1 and summary.duplicate_deliveries == 1
    floated = dict(good, event_id="evt_1004", attempt_number=1.0)
    flagged = dict(good, event_id="evt_9", attempt_number=True)
    empty, bad = review_mappings([good, floated, flagged])
    assert empty == ()
    assert violation_codes(bad)["BAD_TYPE"] == 2
    assert type(True) is bool
    assert isinstance(True, int)  # the trap: a subclass check would accept JSON true
    print(f"accepted={summary.accepted} duplicates={summary.duplicate_deliveries}")


if __name__ == "__main__":
    if len(sys.argv) == 1:
        _demo()
    else:
        with open(sys.argv[1], encoding="utf-8") as handle:
            loaded = load_ndjson(handle)
        _, result = review_mappings(loaded)
        print(f"ok={result.ok} seen={result.seen} codes={dict(violation_codes(result))}")
        sys.exit(0 if result.ok else 2)

Run python3 progress_ndjson.py with no arguments for the demo, which prints accepted=1 duplicates=1. Run it on the sample file and expect ok=False and a non-zero exit status. The float line and the conflicting third line are both enough to reject. The boolean case is in the demo rather than the file because JSON true is easy to miss in a fixture and easy to accept with isinstance.

Tests you should add before calling this done

A table of inputs, not a printout you inspect once:

• Duplicate identical lines yield one event and a duplicate count of one.
• Two bodies, one id, yield ID_CONFLICT and zero accepted events.
• attempt_number values 1.0, true, and "1" each yield BAD_TYPE.
• A naive timestamp yields BAD_TIME.
• A +05:30 offset converts to the same UTC instant as the matching Z time.
• A key contact_email yields PROHIBITED_FIELD and the message does not contain the address.
• learner_id of None does not become the string "None".

Lesson 5 will put these in a test module and run them with unittest or pytest. Writing them now, even as asserts in _demo, is how the coercion bug stays dead.

What this file will not do

It will not insert into PostgreSQL. It will not compute completions. It will not retry a network call. It holds the whole file in a list, which is the limitation lesson 5 removes. If you "improve" it by skipping bad rows, you have changed the batch policy without changing the spec, which is the role defect from lesson 3.

Enterprise Real-World Scenarios

The coerced attempt number

The extract was not the player. A well-meaning export from a spreadsheet tool rewrote integers as floats when an analyst rebuilt a "small" backfill file by hand. The old checker called int() on anything that looked numeric, so 1.0 became attempt 1 and the string "01" became attempt 1 after a different code path stripped zeros. Two learner histories merged. The mart's primary key did not notice, because both became (learner, lesson, 1).

The repair was the strict parser, a rejected backfill, and a rule that hand-built files go through the same command as extract files. Product engineering was consulted and confirmed the player only emits JSON integers. Governance was informed because the spreadsheet had also contained a column named email that the old int() path never looked at; the new prohibited-key loop failed the file for that reason first. The data engineer owned the checker. The analyst owned the unauthorized backfill path, which was then removed. Nobody changed the completion definition. The incident was a parser incident, and the RACI kept it there.

A module cycle that shipped a metric into the parser

A second squad, under schedule pressure, imported the completion function into contract.py so they could "drop incomplete-looking rows early." The completion function imported the catalog client to learn how many checkpoints a lesson has. The contract module therefore opened a network connection when tests imported it, and a catalog outage started failing ingestion. Completeness of the player, which sometimes emits completed without a final checkpoint on old builds, was now a reason to drop raw events. Definition v3 could not be replayed, because the raw zone had been pre-filtered by a later definition.

The repair was the import diagram: contract parses, it does not decide completion, and it does not import the catalog. The early filter was deleted. The days it had touched were re-extracted from the replica, which still had the events, because the operational store was the system of record. The raw zone was not. That sentence from lesson 1 is what made the repair possible. The Python habit that endangered it was a convenient import.

Best Practices

• Parse JSON with the standard library and then check type(value) is int, not isinstance(value, int) and not int(value).
• Represent accepted records as frozen dataclasses. Return tuples from review functions.
• Raise a ContractError with a stable code. Catch that type per row. Let bugs traceback.
• Never use a mutable default argument. Create lists and dicts inside the function.
• Reject whitespace-padded identifiers instead of stripping them.
• Convert aware datetimes to UTC. Reject naive ones. Require Python 3.11+ if you depend on fromisoformat accepting Z.
• Keep import arrows pointed at the contract module. Do not import clients, metrics, or file paths from it.
• Put identifiers and prohibited-field names in the code as data, in a frozenset, not as a chain of string contains checks scattered through the file.
• Exit non-zero on a rejected batch from the command entrypoint only. Libraries raise.
• Prefer a type checker on the public functions. Do not annotate the JSON boundary as Any to make the checker quiet.
• Keep telemetry counts separate from business metrics. The summary is not a completion rate.
• When a batch fits in memory only in development, say so in the module docstring so lesson 5's streaming change is a design, not a surprise.

Common Mistakes and Anti-Patterns

isinstance(x, int) as a JSON integer check. Booleans pass, and True becomes attempt 1 in any arithmetic. The repair is type(x) is int. Add a test with JSON true. This bug is common because the language's subtyping is surprising, not because the author skipped "the basics."

Silent coercion. int(float("1.0")), str(value or ""), and datetime.fromisoformat(text.replace("Z", "")) all produce a row that looks valid and is not the row the producer sent. The repair is to reject and to put the migration, if one is truly required, in an explicit, versioned function that is not on the daily path.

Mutable defaults and module-level accumulators. A list on the function signature or a global SEEN = {} dedupes across files in one process and leaks ids. The repair is local state inside the call, created each time. Long-running workers make this a cross-tenant defect if you ever process two customers in one process.

Catching Exception per row. A misspelled attribute becomes a quarantine increment, the batch is rejected for a reason you cannot find, or, if you skip errors, the batch is accepted without those rows. The repair is ContractError only. Bugs fail the task loudly.

Stripping and lowercasing identifiers. "Lrn_1042" and "lrn_1042" are different until a canonical form is part of the event contract the player also implements. The pipeline does not invent the canonical form. The repair is byte-for-string equality with the operational id, padding rejected.

Business logic in the parser. Dropping events that "could not be completions" destroys replay. The repair is the module boundary. The parser's success means "safe to store," not "counts toward the rate."

Notebook as the production parser. A notebook re-run out of order, with a cell that coerces types still above the strict cell, will reload a day the module would have rejected. The repair is a command someone else can run, with tests, imported by the orchestrator later. Notebooks stay scratch.

Troubleshooting and Design Review

Symptom: Attempt numbers in the raw zone are sometimes one when the player sent a boolean or a float.

Likely cause: a coercion path or isinstance(..., int).

Investigation: read the parser. Search for int(, isinstance, and or "". Reproduce with a one-line NDJSON fixture containing true and 1.0.

Corrective action: deploy the strict check, reject the affected batches, and re-extract from the replica. Do not update the bad rows in place. They are not a reliable record of what arrived.

Symptom: Tests pass and a production file with Z timestamps fails, or the reverse.

Likely cause: the test machine and the worker do not share a Python version, and fromisoformat behavior differs before 3.11. Alternatively the test replaces Z and production does not.

Investigation: print sys.version in the worker. Run the same fixture both places.

Corrective action: pin the runtime to 3.11 or newer in the job definition. Delete the replace hack so there is one parser.

Symptom: A rejected batch's message contains a learner email address.

Likely cause: the exception formatter included repr(payload).

Investigation: read ContractError construction. Search for repr(payload) and f-strings that interpolate values of prohibited keys.

Corrective action: log the key name and the line number. Rotate any log store that retained the address. Add a test that a payload containing secret@example.com produces a message without that substring.

Symptom: Importing the contract module opens a database connection.

Likely cause: an import cycle or a side effect at module level, usually a client constructed at import so the metric can "validate early."

Investigation: run python -c "import contract" with the network disabled and see what fails. Draw the import graph.

Corrective action: move the client to the job that actually publishes. The contract module imports nothing but the standard library.

Symptom: Memory grows file over file in a worker that calls the review in a loop.

Likely cause: a mutable default, a global dict, or a summary list the caller keeps appending to the same object.

Investigation: look for =[] and module-level collections. Log len of the index at the start of each file. It should be zero.

Corrective action: construct collectors inside the function. Add a test that reviews two batches in one process and checks that an id from the first is not "already seen" in the second.

A design review of pipeline Python asks: where is coercion possible, which exception type is a bad record, what is frozen, which module is allowed to do IO, and what happens to a conflicting event_id? If the answer to the last question is "we keep the latest," the reviewer asks who defined latest and why the batch policy from lesson 1 was changed in a helper function.

Interview Questions and Answers

Q1. Why is isinstance(attempt, int) the wrong check for a JSON integer in a data pipeline?

Because bool is a subclass of int in Python. JSON true becomes True, isinstance(True, int) is true, and True == 1. A progress event can then occupy attempt 1 without the producer ever sending an attempt number. I check type(attempt) is int and I add a fixture with true. I also reject floats such as 1.0 even though int(1.0) would succeed. Coercion hides a producer or a spreadsheet rewrite. The raw zone should store the integer the contract allowed, or nothing.

Q2. What is wrong with a mutable default argument in a long-running ingest worker?

The default list or dict is created once when the function is defined and then shared by every call that omits the argument. A worker that processes file after file will accumulate violations, seen ids, or learner identifiers from earlier files. That is a correctness bug and, if the values are personal, a data leak across batches. I create the collection inside the function. I do not "clear it at the end," because an exception will skip the clear.

Q3. How do you separate a bad record from a bug in the parser?

Bad records raise a dedicated ContractError with a stable code. The row loop catches that type, records it, and applies the batch policy. Attribute errors, name errors, and unexpected type errors propagate so the job fails as a defect in our code, which is a different severity from a producer sending a float. I do not catch Exception. I do not convert every ValueError in a large try that also contains business logic, because I will mislabel my own mistakes as quarantine.

Q4. When would you use a tuple, a frozen dataclass, or a dict for a progress event?

A dict is the untrusted JSON object and nothing further. A frozen dataclass is the accepted record: named fields, equality by payload, hashable, no assignment. A tuple is appropriate for a small key or for an immutable sequence of events returned to the caller. I do not pass a five-field tuple through the pipeline, because the positions will be swapped under maintenance. I do not keep the raw dict "because we might need extra keys." Extra keys are either part of the contract or a reason to reject, especially if they are prohibited.

Q5. How should duplicate event_id values be represented in Python, not just in prose?

An index from id to the frozen event. If the new event equals the stored one, increment a duplicate-delivery count and do not append. If it does not equal, append a violation with code ID_CONFLICT and accept nothing from that batch. Equality on the dataclass is field equality, not id equality. If __eq__ used only the id, conflicts would compare equal and vanish. I do not use a set of ids alone, because a set cannot show that the payload differed.

Q6. Where do type hints help a data pipeline, and where do they give false confidence?

They document and check the trusted side of the boundary: functions that already hold a ProgressEvent, summaries, and module interfaces. A checker will catch a caller that passes a string where an event is required, if the annotations are honest and Any is not smeared over them. They do not validate JSON. json.loads returns a dynamic object, and I narrow it with runtime checks. Annotations on parse_event do not reject an email key. The prohibited-key loop does. I still want the hints, and I do not treat a green checker as a green contract test.

Q7. How do you handle timestamps in the ingest function?

I require a string, parse it with datetime.fromisoformat on Python 3.11 or newer so Z is legal, reject a naive result, and convert to UTC with astimezone. I do not drop the offset and I do not apply the business timezone in the parser. "Which calendar day is Tuesday for the metric" is a spec decision applied when the mart is built, in SQL, where the definition lives. Mixing display zones into the raw event makes two files for the same instant look different.

Q8. Why is payload.get("lesson_id") or "" unacceptable?

It collapses missing, null, and empty into one value, so the quarantine cannot tell a producer bug from a default we invented. It also treats any future legitimate value that is falsy as missing. For strings, a whitespace id would pass a later truthiness test if I only checked if lesson_id. I want _require_str to accept only a non-empty str with no padding, and to raise ContractError otherwise. The raw zone then does not contain a learner whose id is the string "None", which is what str(None) produces if someone "fixed" a type error that way.

Q9. How do you structure modules for a contract checker?

The contract module imports only the standard library and knows fields and types. The batch module applies uniqueness and the accept-or-reject policy. The NDJSON module knows framing. The command module knows paths and exit codes. Imports point toward the contract, never from the contract toward a database, a metric, or a path. I keep sys.exit out of library code. This is the same ownership idea as the RACI: the parser owner should not be able to change completion by editing a helper, and the import graph is what enforces it between code reviews.

Q10. A colleague wants to filter out events that are not completions inside the parser, to save storage. What do you say?

I refuse. The parser's job is to decide whether the batch is safe to store, not whether it counts toward a metric. Early filtering destroys replay when the definition changes, and it pulls catalog knowledge and metric code into the ingest path, which is how a catalog outage stops ingestion. Storage of contracted events for the retention window is the cost of being able to rebuild Tuesday. If the volume is actually too high, we partition and compress, or we revisit retention. We do not drop facts because today's definition does not count them.

Q11. When is Python the wrong tool for the next step after a file is accepted?

When the next step is a join to enrollment, a distinct count, or a replace-by-day at a relational grain. That work belongs in SQL on the curated table, where the consumer can read the same statement the job ran. Reimplementing it in nested dicts duplicates the definition and is harder to review. Python remains the right tool for the untrusted boundary, for files, and for rules that must fail a batch before a database sees it. Lesson 6 takes the accepted shape and writes the set-based query. I do not extend this script into a warehouse.

Q12. What tests do you consider mandatory before this parser is allowed to reject a production day?

A duplicate identical pair, a conflicting pair, JSON true and 1.0 and "1" for the attempt, a naive timestamp, an offset that must convert to UTC, a prohibited key whose value must not appear in the error text, a null identifier that must not become the string "None", and a second batch in the same process that must not see the first batch's ids. I also want a framing test for a blank line and for a non-object JSON value. I do not consider a successful print of one happy-path file a test suite. The coercion bugs this lesson is about are invisible on the happy path.

Hands-On Exercises

Exercise: Tighten a progress-event parser

Problem statement. You are given a directory of NDJSON files that were accepted by a coercive checker. Implement the strict parser from this lesson, without copying the demo assertions as your only tests, and report what the old files would have distorted.

Tasks.

1. Implement parse_event, review_mappings, and load_ndjson with the types and errors specified above. Use Python 3.11+ and the standard library.
2. Add tests for the boolean, float, string, duplicate, conflict, naive time, offset, prohibited field, and cross-batch cases.
3. Include one file that is valid and one that must exit with status 2.
4. Show that isinstance(True, int) is true and that your parser still rejects JSON true. Explain in two sentences why the test exists.
5. Ensure error messages for a prohibited field do not contain the field's value.
6. Draw the module import graph you would use if you split the file, and confirm the contract module does not import IO.
7. State one rule you refused to implement in the parser even though it would "clean up" the mart, and name the lesson that owns that rule.
8. Note the memory limitation and what lesson 5 must change before a peak-day file is safe.

Expected deliverables. The module, the tests and their output, one rejected fixture, the import note, and the limitation note.

Evaluation rubric.

You pass when a reviewer can point at the line that rejects true and at the test that would fail if someone "simplified" it to isinstance.

Lesson Summary

Python in a data pipeline is a contract language before it is a convenience language. The standard types will accept a boolean as an integer, share a default list across a worker's lifetime, and stringify None into an identifier if you help them. The raw zone will then remember your helpfulness as history.

The habits that prevent that are strict type checks, frozen records, pure review functions, a dedicated error type, UTC-aware timestamps, and modules whose imports point at the contract rather than at the metric. The summary counts you print are about the file, not about learner completion. SQL will own completion. Generators, context managers, and a real test layout will own the next lesson, because a list that holds the day is still a teaching constraint, and the strictness you added here is what must survive when the file no longer fits in memory.

Further Reading

• Python 3 documentation: built-in types — numeric types, the fact that bool is a subclass of int, and mutable sequence behavior.
• Python 3 documentation: datetime — aware versus naive objects, and fromisoformat accepting Z from 3.11.
• Python 3 documentation: json — what loads returns for objects, numbers, and true.
• Python 3 documentation: dataclasses — frozen=True, equality, and hashing.
• PEP 8 — layout. It will not save a bad contract, and it will keep the module readable enough to review one.
• The language primer /learn/python — syntax this lesson assumed. Use it for gaps. Use this lesson for the pipeline boundary.
• Continue this course: /learn/data-engineering/advanced-python-for-data-pipelines, then /learn/data-engineering/sql-fundamentals-for-data-engineers.