Lab 1 — Build an ingestion script

Objective

By the end of this lab you will have a runnable Python script, ingest.py, that reads raw order records from a CSV file and a JSON file, validates every row, writes the clean records as JSONL, and logs a summary of how many rows it read, kept, and dropped. It is the first stage of a real pipeline — the part that turns messy source files into trustworthy input for everything downstream — and it uses only the Python standard library (csv, json, logging, pathlib), so it costs nothing and runs anywhere.

Skills exercised:

Reading CSV and JSON sources with streaming generators
Validating raw rows into clean, typed records
Writing clean output as JSON Lines
Logging a skip-and-count run summary

Prerequisites and setup

Prerequisites: the three lessons in this module — Data structures for pipelines, Reading and writing data formats, and Errors, logging, and robust scripts.

Setup: you need Python 3.10 or newer. There are no packages to install; everything uses the standard library. Verify your version and create a working folder.

macOS / Linux
Windows (PowerShell)

python3 --version
mkdir de103-ingestion-lab && cd de103-ingestion-lab

python --version
mkdir de103-ingestion-lab; cd de103-ingestion-lab

Expected output from the version command is a line like Python 3.11.6. If the command is not found, install Python from python.org before continuing. In the steps below, use python3 on macOS/Linux and python on Windows.

note

Every step writes or edits files in this one folder. If you close your terminal and come back, re-run the cd de103-ingestion-lab command to return to it; all your files are still there.

Step 1 — Create the raw source files

Create the two raw inputs your script will ingest: a CSV export and a JSON export, each with the same order schema and some deliberate defects.

Save this exactly as raw_orders.csv:

order_id,event_time,category,amount,status
5001,2026-07-16T08:00:00,books,20.00,paid
5002,2026-07-16T09:15:00,electronics,299.00,paid
5003,2026-07-16T10:30:00,books,,paid
5004,2026-07-16T11:00:00,electronics,150.00,refunded

Save this exactly as raw_orders.json:

[
  {
    "order_id": "6001",
    "event_time": "2026-07-16T12:00:00",
    "category": "toys",
    "amount": "45.50",
    "status": "paid"
  },
  {
    "order_id": "6002",
    "event_time": "2026-07-16T13:00:00",
    "category": "toys",
    "amount": "-5.00",
    "status": "paid"
  },
  {
    "order_id": "",
    "event_time": "2026-07-16T14:00:00",
    "category": "books",
    "amount": "10.00",
    "status": "paid"
  }
]

These files carry defects on purpose: order 5003 is missing its amount, 6002 has a negative amount, and the last JSON record has an empty order_id. Your script will drop all three.

Checkpoint: confirm both files exist and are readable.

python3 -c "import csv, json; print(len(list(csv.DictReader(open('raw_orders.csv')))), len(json.load(open('raw_orders.json'))))"

You should see 4 3 — four CSV data rows and three JSON records. If you see a FileNotFoundError, check that both files are saved in the current folder with exactly those names.

Step 2 — Set up logging and the source readers

Goal: create ingest.py with logging configured and two streaming readers, one per format. Each reader yields raw dicts and tags them with their source.

Create ingest.py with this content:

import csv
import json
import logging
from datetime import datetime
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("ingest")

REQUIRED_FIELDS = ("order_id", "event_time", "category", "amount")


def read_csv(path):
    """Yield raw records from a CSV file, tagging each with its source."""
    with open(path, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            row["source"] = "csv"
            yield row


def read_json(path):
    """Yield raw records from a JSON array file, tagging the source."""
    with open(path, encoding="utf-8") as f:
        for row in json.load(f):
            row["source"] = "json"
            yield row

Checkpoint: count the raw rows each reader streams.

python3 -c "import ingest; from pathlib import Path; \
c = sum(1 for _ in ingest.read_csv(Path('raw_orders.csv'))); \
j = sum(1 for _ in ingest.read_json(Path('raw_orders.json'))); \
print(c, j, c + j)"

You should see 4 3 7. If you get a ModuleNotFoundError: No module named 'ingest', you are running from the wrong folder — cd into de103-ingestion-lab where ingest.py lives.

Step 3 — Validate a raw row into a clean record

Goal: add the rule that decides whether a raw row is good. validate returns a clean record dict, or None (after logging a warning) when the row breaks a rule.

Add this function to ingest.py:

def validate(row):
    """Return a clean record dict, or None if the row breaks a rule."""
    order_id = row.get("order_id")
    for field in REQUIRED_FIELDS:
        if not row.get(field):
            logger.warning("drop %r: missing %s", order_id, field)
            return None
    try:
        event_time = datetime.fromisoformat(row["event_time"])
    except ValueError:
        logger.warning("drop %r: bad event_time", order_id)
        return None
    try:
        amount = float(row["amount"])
    except (TypeError, ValueError):
        logger.warning("drop %r: bad amount %r", order_id, row["amount"])
        return None
    if amount < 0:
        logger.warning("drop %r: negative amount %s", order_id, amount)
        return None
    return {
        "order_id": str(order_id),
        "event_time": event_time.isoformat(),
        "category": row["category"],
        "amount": amount,
        "status": row.get("status", ""),
        "source": row["source"],
    }

Checkpoint: validate one good row and one bad row.

python3 -c "import ingest; \
good = {'order_id':'9001','event_time':'2026-07-16T00:00:00','category':'books','amount':'12.00','status':'paid','source':'csv'}; \
bad = dict(good, order_id='9002', amount=''); \
print(ingest.validate(good)); print(ingest.validate(bad))"

You should see a clean dict for the good row (with amount as the float 12.0), then a WARNING drop '9002': missing amount line, then None. Seeing the warning here means the guard works — that is the intended result, not a failure.

Step 4 — Write clean JSONL and assemble the pipeline

Goal: add a JSONL writer and the ingest orchestrator that streams every source through validate, writes the survivors, and logs a run summary.

Add these two functions to ingest.py:

def write_jsonl(records, path):
    """Write clean records as JSON Lines; return the count written."""
    written = 0
    with open(path, "w", encoding="utf-8") as f:
        for record in records:
            f.write(json.dumps(record) + "\n")
            written += 1
    return written


def ingest(sources, out_path):
    """Read every source, validate, write clean JSONL, log a summary."""
    stats = {"read": 0, "valid": 0, "invalid": 0}

    def clean_records():
        for source in sources:
            reader = read_csv(source) if source.suffix == ".csv" else read_json(source)
            for row in reader:
                stats["read"] += 1
                record = validate(row)
                if record is None:
                    stats["invalid"] += 1
                    continue
                stats["valid"] += 1
                yield record

    written = write_jsonl(clean_records(), out_path)
    stats["written"] = written
    logger.info(
        "summary: read=%d valid=%d invalid=%d written=%d",
        stats["read"], stats["valid"], stats["invalid"], written,
    )
    return stats

Checkpoint: confirm the writer works on a throwaway list.

python3 -c "import ingest; \
n = ingest.write_jsonl([{'order_id': '1', 'amount': 1.0}], 'sample.jsonl'); \
print('wrote', n)"

You should see wrote 1, and a sample.jsonl file appears in the folder. You can delete it — it was only a smoke test.

Step 5 — Run the script end to end

Goal: add the entry point that wires the real sources together, then run the whole ingestion and inspect its output.

Add this to the bottom of ingest.py:

if __name__ == "__main__":
    base = Path(__file__).parent
    sources = [base / "raw_orders.csv", base / "raw_orders.json"]
    ingest(sources, base / "clean_orders.jsonl")

Checkpoint: run the script, then look at the clean output.

python3 ingest.py
cat clean_orders.jsonl

The run logs three WARNING drop ... lines (for 5003, 6002, and the empty order_id) and ends with INFO summary: read=7 valid=4 invalid=3 written=4. The clean_orders.jsonl file then holds exactly four lines — orders 5001, 5002, 5004, and 6001 — each a JSON object with amount as a number. On Windows PowerShell, use Get-Content clean_orders.jsonl instead of cat.

Validation

Verify the whole build against the objective: clean JSONL that is correct end to end. Save this as validate.py:

import json
from pathlib import Path

EXPECTED_IDS = ["5001", "5002", "5004", "6001"]
EXPECTED_TOTAL = 514.5

lines = Path("clean_orders.jsonl").read_text(encoding="utf-8").splitlines()
records = [json.loads(line) for line in lines]
ids = [r["order_id"] for r in records]
total = round(sum(r["amount"] for r in records), 2)

if ids == EXPECTED_IDS and total == EXPECTED_TOTAL:
    print("PASSED — ingestion produced 4 clean rows totaling 514.50")
else:
    print("FAILED — got ids=%s total=%s" % (ids, total))

Run it:

python3 validate.py

Checkpoint: the command prints PASSED — ingestion produced 4 clean rows totaling 514.50. If it prints FAILED, open clean_orders.jsonl: it must have exactly the four order ids 5001, 5002, 5004, 6001, and the amount values must sum to 514.50. Re-run python3 ingest.py first if you edited the raw files.

Teardown

Nothing is billable — this lab runs entirely on your machine. Keep the de103-ingestion-lab folder if you want to reuse ingest.py as a reference for the course lab challenges; otherwise remove it with rm -rf de103-ingestion-lab (macOS/Linux) or Remove-Item -Recurse de103-ingestion-lab (Windows). You can also delete the throwaway sample.jsonl from Step 4 at any time.

Troubleshooting

Symptom	Likely cause	Fix
`ModuleNotFoundError: No module named 'ingest'`	Running from the wrong folder	`cd` into `de103-ingestion-lab`, where `ingest.py` lives
`FileNotFoundError: 'raw_orders.csv'`	Source file missing or misnamed	Re-save the Step 1 files with exactly those names in the folder
Checkpoint prints `3 3 6` in Step 2	CSV missing the `5004` row	Re-save `raw_orders.csv` with all four data rows from Step 1
Summary shows `written=7` in Step 5	`validate` never returns `None`	Confirm each guard in Step 3 returns `None`, not the row
Summary shows `written=3` in Step 5	A valid row is being dropped	Check the negative-amount guard uses `amount < 0`, not `<=`
Validation prints `FAILED`	Stale or edited output file	Re-run `python3 ingest.py`, then `python3 validate.py` again

Objective​

Prerequisites and setup​

Step 1 — Create the raw source files​

Step 2 — Set up logging and the source readers​

Step 3 — Validate a raw row into a clean record​

Step 4 — Write clean JSONL and assemble the pipeline​

Step 5 — Run the script end to end​

Validation​

Teardown​

Troubleshooting​