Build a Fault-Tolerant Python ETL

Introduction

30 Second Summary

Online stores depend on every valid order reaching fulfilment. One damaged record can interrupt that flow or disappear without enough evidence to diagnose the problem.

In this project, you will build a command-line e-commerce order-ingestion program using the Python standard library. The finished pipeline preserves valid sales through a fault-tolerant ETL run with visible failure records.

What You'll Build

You will demo a damaged order feed that still produces valid accepted orders plus inspectable evidence for every rejection.

By the end of this project, you'll have:

• A runnable reference-data check that proves the CSV product catalog plus the JSON customer document loaded correctly.
• A strict streaming run that writes the first valid order before exposing a malformed JSONL record through a chained custom exception.
• A resilient ETL run you can inspect through its UTF-8 accepted-order file, rejection quarantine, plus final run summary.
• Secret Mission: corrupt the incoming feed at the byte level to prove that source encoding failures need a separate boundary from record validation failures.

Are there any prerequisites?

You need Visual Studio Code on Linux plus familiarity with Python functions, loops, dictionaries, classes, plus command-line script execution. The lab targets CPython 3.10 using only its standard library.

Before We Start

Before you begin building, lock in what this e-commerce order-ingestion pipeline must protect. Your answer gives every later decision about failures a clear business reason.

Set Up and Seed the Interview Lab

A fault-tolerant ETL interview answer becomes convincing when every failure comes from reproducible source data. This lab gives each later error a specific record you can trace.

You will prepare a Visual Studio Code workspace around a Python 3.10-compatible seeder. You will also verify the interpreter that runs every later check.

In this step, get ready to:

• Verify the available Python command plus its version.
• Prepare the `ecommerce-order-etl` workspace with a repeatable data seeder.
• Inspect the generated file formats plus their deliberate data-quality problems.

Prepare the workspace and verify Python

The interpreter version defines which language features your scripts can use. This project stays within the Python 3.10 standard-library surface.

• Use your Linux application launcher to start Visual Studio Code.
• Use the folder controls in Visual Studio Code to create `ecommerce-order-etl` on your Desktop.
• Select `ecommerce-order-etl` as the active workspace folder.
• Open the integrated terminal from the Visual Studio Code terminal controls.
• Check the available Python version by running this command:

python3 --version

What does this check prove?

The command prints the Python version attached to python3. The remaining scripts use that same command unless your machine exposes a different executable.

✔️ I see a Python 3 version

Your interpreter is available for the project. The code avoids features introduced after Python 3.10.

• Record the reported version here: your Python version.

ⓧ Command not found

Your machine may expose Python through a different executable. Use the Python executable already configured on the machine.

• Replace python3 in the version command with your configured Python executable.
• Run the updated version command.
• Substitute that executable for python3 in every remaining command.

Need help finding Python?

Use the interpreter already configured for Python scripts on your Linux machine. Help me identify the Python executable configured on my Linux machine.

Build the data seeder

The seed_data.py script creates a CSV catalog. It creates one complete JSON customer document.

The script also creates a JSONL order feed. Explicit UTF-8 handling preserves names such as José plus product text such as Café.

• Use the file creation control in the Visual Studio Code Explorer sidebar to create seed_data.py inside ecommerce-order-etl.
• Initialize the data directory plus the catalog fixtures by pasting this first chunk:

import csv
import json
from pathlib import Path


DATA_DIR = Path("data")
DATA_DIR.mkdir(parents=True, exist_ok=True)

catalog_rows = [
    {
        "sku": "SKU-100",
        "product_name": "Mechanical Keyboard",
        "unit_price_cents": 7900,
    },
    {
        "sku": "SKU-200",
        "product_name": "Café Mug",
        "unit_price_cents": 4500,
    },
]

What does this code prepare?

• The pathlib import provides Path objects for file locations.
• DATA_DIR points to the local data directory.
• catalog_rows holds two product records with prices stored as integer cents.

• Save seed_data.py.
• Run seed_data.py with the editor's Python run control.

You should see a data folder appear in the Explorer sidebar. The terminal should return to its prompt without a traceback.

Data folder missing?

Confirm that seed_data.py is saved inside ecommerce-order-etl. Check that the editor used the Python interpreter you verified earlier.

Still stuck? Help me find why seed_data.py does not create its data folder.

• Add the catalog writer directly below catalog_rows by pasting this code:

with (DATA_DIR / "catalog.csv").open(
    "w", encoding="utf-8", newline=""
) as catalog_file:
    writer = csv.DictWriter(
        catalog_file,
        fieldnames=["sku", "product_name", "unit_price_cents"],
    )
    writer.writeheader()
    writer.writerows(catalog_rows)

How is the catalog written?

• The file opens in text write mode with explicit UTF-8 encoding.
• The empty newline value leaves newline handling to the CSV module.
• DictWriter writes a fixed header before the two catalog rows.

• Save seed_data.py.
• Run seed_data.py with the editor's Python run control.
• Select data/catalog.csv in the Explorer sidebar.

You should see the three-column header. You should also see rows for SKU-100 plus SKU-200.

Catalog file missing?

Check that the catalog writer sits after the closing bracket for catalog_rows. Confirm that the indentation inside the with block matches the snippet.

Need another set of eyes? Help me troubleshoot why catalog.csv is not being generated.

• Add the customer fixtures directly below the catalog writer by pasting this code:

customers = [
    {"customer_id": "C-001", "name": "Amina Yusuf"},
    {"customer_id": "C-002", "name": "José Silva"},
]

with (DATA_DIR / "customers.json").open(
    "w", encoding="utf-8"
) as customers_file:
    json.dump(customers, customers_file, ensure_ascii=False, indent=2)
    customers_file.write("\n")

How is the customer document written?

• customers holds two customer objects in one list.
• json.dump serializes that list as one complete document.
• ensure_ascii=False preserves José as readable UTF-8 text.

• Save seed_data.py.
• Run seed_data.py with the editor's Python run control.
• Select data/customers.json in the Explorer sidebar.

You should see one JSON array containing Amina Yusuf plus José Silva. The accented character should remain readable.

Customer JSON missing?

Confirm that the customer writer uses customers_file in both write calls. Check the closing bracket after the second customer object.

Need help with the structure? Help me find the syntax problem in my customer JSON seeding code.

• Add the order fixtures directly below the customer writer by pasting this code:

orders = [
    {
        "order_id": "O-1001",
        "customer_id": "C-001",
        "sku": "SKU-100",
        "quantity": 1,
    },
    {
        "order_id": "O-1003",
        "customer_id": "C-001",
        "sku": "SKU-999",
        "quantity": 1,
    },
    {
        "order_id": "O-1004",
        "customer_id": "C-002",
        "sku": "SKU-200",
        "quantity": "2",
    },
]

order_lines = [
    json.dumps(orders[0], ensure_ascii=False),
    '{"order_id":"O-1002","customer_id":"C-002","sku":"SKU-200","quantity":2',
    json.dumps(orders[1], ensure_ascii=False),
    json.dumps(orders[2], ensure_ascii=False),
]

What problems are seeded here?

• The first order uses known customer plus product identifiers.
• The second line deliberately omits its closing brace.
• The third order uses the unknown SKU SKU-999.
• The fourth order stores quantity as the string 2.

• Save seed_data.py.
• Run seed_data.py with the editor's Python run control.

The terminal should return to its prompt without a traceback. This proves the fixtures are valid Python values before the script writes the JSONL feed.

Seeing a Python syntax error?

The malformed order is intentionally stored as a quoted Python string. Confirm that its opening plus closing single quotes match the snippet.

Still seeing an error? Help me distinguish the intentional malformed JSON string from a Python syntax error.

• Finish the seeder directly below order_lines by pasting this code:

(DATA_DIR / "orders.jsonl").write_text(
    "\n".join(order_lines) + "\n",
    encoding="utf-8",
)

print("Seeded data/catalog.csv, data/customers.json, and data/orders.jsonl")

How is the feed finished?

• The newline join places one order value on each source line.
• write_text saves the complete feed with UTF-8 encoding.
• The final print gives the run a concrete success signal.

• Save seed_data.py.

Before you run the completed script, which three files do you expect to appear inside data?

• Generate the complete fixture set from the integrated terminal by running:

python3 seed_data.py

What does this command run?

Python executes seed_data.py from the active ecommerce-order-etl workspace. The script rewrites the three source files into a known starting state.

• Confirm that the terminal prints Seeded data/catalog.csv, data/customers.json, and data/orders.jsonl.
• Expand the data folder in the Explorer sidebar.
• Confirm that catalog.csv, customers.json, plus orders.jsonl are listed.

Seeder command failed?

Confirm that the terminal path ends in ecommerce-order-etl. Check that seed_data.py is saved before rerunning the command.

Need help reading the traceback? Help me troubleshoot my seed_data.py execution error.

Use this comparison if you want to check the completed seeder before moving on.

✔️ Awesome, I've got everything!

Great work. Save seed_data.py once more so the workspace matches the generated files.

ⓧ I'd like to double check the full code

import csv
import json
from pathlib import Path


DATA_DIR = Path("data")
DATA_DIR.mkdir(parents=True, exist_ok=True)

catalog_rows = [
    {
        "sku": "SKU-100",
        "product_name": "Mechanical Keyboard",
        "unit_price_cents": 7900,
    },
    {
        "sku": "SKU-200",
        "product_name": "Café Mug",
        "unit_price_cents": 4500,
    },
]

with (DATA_DIR / "catalog.csv").open(
    "w", encoding="utf-8", newline=""
) as catalog_file:
    writer = csv.DictWriter(
        catalog_file,
        fieldnames=["sku", "product_name", "unit_price_cents"],
    )
    writer.writeheader()
    writer.writerows(catalog_rows)

customers = [
    {"customer_id": "C-001", "name": "Amina Yusuf"},
    {"customer_id": "C-002", "name": "José Silva"},
]

with (DATA_DIR / "customers.json").open(
    "w", encoding="utf-8"
) as customers_file:
    json.dump(customers, customers_file, ensure_ascii=False, indent=2)
    customers_file.write("\n")

orders = [
    {
        "order_id": "O-1001",
        "customer_id": "C-001",
        "sku": "SKU-100",
        "quantity": 1,
    },
    {
        "order_id": "O-1003",
        "customer_id": "C-001",
        "sku": "SKU-999",
        "quantity": 1,
    },
    {
        "order_id": "O-1004",
        "customer_id": "C-002",
        "sku": "SKU-200",
        "quantity": "2",
    },
]

order_lines = [
    json.dumps(orders[0], ensure_ascii=False),
    '{"order_id":"O-1002","customer_id":"C-002","sku":"SKU-200","quantity":2',
    json.dumps(orders[1], ensure_ascii=False),
    json.dumps(orders[2], ensure_ascii=False),
]

(DATA_DIR / "orders.jsonl").write_text(
    "\n".join(order_lines) + "\n",
    encoding="utf-8",
)

print("Seeded data/catalog.csv, data/customers.json, and data/orders.jsonl")

What should match?

Compare the imports plus each data-writing section with your saved file. Matching the order matters because each section uses values defined above it.

Inspect the formats and rehearse the file-handling answers

Each source format represents a different file-handling decision. Their visible structure gives you an interview-ready reason for choosing each one.

• Select data/catalog.csv in the Explorer sidebar.
• Identify its header plus two product rows as tabular CSV data.
• Select data/customers.json in the Explorer sidebar.
• Identify its surrounding array as one complete JSON document.
• Select data/orders.jsonl in the Explorer sidebar.
• Compare the four independent lines in the order feed.

Line 1 contains a valid order. Line 2 contains malformed JSON.

Line 3 contains the unknown SKU SKU-999. Line 4 contains a string quantity.

• Answer aloud: What is the difference between text mode plus binary mode?
• Answer aloud: Why use a context manager for files?
• Sketch the smallest file loop that processes one line at a time.

Interview model answers

• Text mode decodes bytes into str. It can translate line endings.
• Binary mode returns bytes. It performs no text decoding.
• A context manager closes the file promptly when processing raises an exception.
• A memory-conscious loop opens the file with a with statement. It iterates directly over the file object one line at a time.

Before the final check, predict whether rerunning the seeder creates duplicates or restores the same fixture files.

• Run the end-of-step verification in the integrated terminal:

python3 seed_data.py

Why is the result repeatable?

Each output opens in write mode. Every run replaces the previous fixture contents with the same known records.

You should see Seeded data/catalog.csv, data/customers.json, and data/orders.jsonl in the terminal. That is your interview lab seeded with traceable failures.

• Confirm that data/catalog.csv contains the header plus two product rows.
• Confirm that data/customers.json contains two customer objects in one array.
• Confirm that data/orders.jsonl contains exactly four nonblank lines.

Generated files look different?

Rerunning the complete saved script should restore every fixture. Compare seed_data.py with the full-code tab if a row or line is missing.

Need help comparing the artifacts? Help me diagnose why my seeded CSV, JSON, or JSONL file differs from the expected fixtures.

Your reproducible source data is ready. Next, you will load the catalog plus customer references into keyed lookups.

Load CSV and JSON Reference Data

Your seeded files now give the pipeline a realistic product catalog and customer list. This step turns those small reference files into lookups that every order can use.

Orders cannot be validated or enriched until the pipeline can trust its reference data. A broken catalog or customer document makes every later result unreliable.

The catalog uses CSV. The customer document uses JSON. Both loaders use explicit UTF-8 handling.

In this step, get ready to:

• Build a catalog lookup from the product CSV file.
• Build a customer lookup from the complete JSON document.
• Run a reference-data check from the command line.

Build the catalog lookup

A reference-data loader checks a source before the order pipeline depends on it. The catalog loader validates its headers before converting each price into integer cents.

Why store prices as integer cents?

Integer cents keep monetary calculations exact in this example. They avoid the binary floating-point rounding that can affect decimal prices.

• Create order_etl.py inside the ecommerce-order-etl folder from the Visual Studio Code Explorer sidebar.
• Add the imports and accepted-order field order by copying this code:

import argparse
import csv
import json
from pathlib import Path


ACCEPTED_FIELDS = [
    "order_id",
    "customer_name",
    "sku",
    "product_name",
    "quantity",
    "total_cents",
]

What does this foundation do?

• The imports provide command-line parsing, CSV handling, JSON handling, and filesystem paths.
• The ACCEPTED_FIELDS list fixes the column order used by the completed pipeline.

• Save order_etl.py.
• Confirm that order_etl.py appears beside seed_data.py in the Explorer sidebar.

File missing from the workspace?

Check that you created order_etl.py inside ecommerce-order-etl. A file created outside that folder does not appear beside the seeded files.

Need help locating it? Help me find order_etl.py in my Visual Studio Code workspace.

• Add the exception hierarchy directly below ACCEPTED_FIELDS by copying this code:

class ETLError(Exception):
    pass


class ReferenceDataError(ETLError):
    pass


class OrderRecordError(ETLError):
    def __init__(self, line_number, message):
        self.line_number = line_number
        super().__init__("Line {}: {}".format(line_number, message))


class OrderDecodeError(OrderRecordError):
    pass


class OrderValidationError(OrderRecordError):
    pass


class SourceEncodingError(ETLError):
    pass

How does the exception hierarchy help?

Each subclass names a different failure boundary. Later processing can catch expected record failures without hiding source failures or programming defects.

• Save order_etl.py.
• Confirm that ReferenceDataError inherits from ETLError.

Seeing a class error?

Keep every class at the left edge of the file. Indent only the body of each class.

Need a second pair of eyes? Help me check the exception hierarchy in order_etl.py.

The first part of load_catalog() opens the catalog with explicit decoding. It checks the header before trusting any rows.

• Add the initial load_catalog() structure below SourceEncodingError by copying this code:

def load_catalog(path):
    catalog = {}

    try:
        with path.open("r", encoding="utf-8", newline="") as handle:
            reader = csv.DictReader(handle)
            required_headers = {"sku", "product_name", "unit_price_cents"}
            actual_headers = set(reader.fieldnames or [])

            if not required_headers.issubset(actual_headers):
                missing = sorted(required_headers.difference(actual_headers))
                raise ReferenceDataError(
                    "Catalog is missing headers: {}".format(", ".join(missing))
                )
    except csv.Error as exc:
        raise ReferenceDataError(
            "Could not parse catalog CSV: {}".format(path)
        ) from exc

    return catalog

What does the header check do?

• The with statement closes the catalog file when processing finishes or fails.
• The encoding="utf-8" argument decodes names such as Café Mug consistently.
• The newline="" argument lets the CSV module perform its own newline handling.
• The required-header set catches a missing column before row processing starts.

• Save order_etl.py.
• Confirm that the required header set contains sku, product_name, and unit_price_cents.

Seeing an indentation problem?

Keep the header check inside the file-opening block. Keep except csv.Error as exc: aligned with try:.

Need help checking the structure? Help me fix the indentation in load_catalog().

Once the header is trustworthy, each row can become a product entry. The SKU becomes the lookup key.

• Place this loop after the required-header check in load_catalog().
• Keep the loop before except csv.Error as exc:.

            for row in reader:
                sku = (row.get("sku") or "").strip()
                product_name = (row.get("product_name") or "").strip()

                try:
                    unit_price_cents = int(row.get("unit_price_cents"))
                except (TypeError, ValueError) as exc:
                    raise ReferenceDataError(
                        "Catalog line {} has an invalid unit_price_cents".format(
                            reader.line_num
                        )
                    ) from exc

                if not sku or not product_name or unit_price_cents < 0:
                    raise ReferenceDataError(
                        "Catalog line {} contains invalid values".format(
                            reader.line_num
                        )
                    )

                catalog[sku] = {
                    "product_name": product_name,
                    "unit_price_cents": unit_price_cents,
                }

How does the catalog lookup form?

• Each csv.DictReader row contains strings keyed by the CSV headers.
• The integer conversion validates unit_price_cents before the value enters the lookup.
• The final dictionary uses each SKU as a key. Product details become available without another CSV scan.

• Save order_etl.py.
• Confirm that the for row in reader: loop remains inside the file-opening block.

Catalog loop in the wrong place?

Align for row in reader: with the preceding header check. Keep the loop nested inside the with block.

Still unsure? Help me place the catalog row loop correctly.

Load customers and add check mode

The customer file contains one complete JSON document. Its small list can be loaded once before the order stream begins.

• Add the initial load_customers() structure below load_catalog() by copying this code:

def load_customers(path):
    try:
        with path.open("r", encoding="utf-8") as handle:
            payload = json.load(handle)
    except json.JSONDecodeError as exc:
        raise ReferenceDataError(
            "Could not parse customer JSON: {}".format(path)
        ) from exc

    if not isinstance(payload, list):
        raise ReferenceDataError("Customer JSON must contain a list")

    customers = {}
    return customers

Why use json.load here?

The json.load function reads one complete JSON document from an open file. A decoder failure becomes ReferenceDataError while exception chaining preserves the original cause.

The list check rejects valid JSON with the wrong business structure. Valid syntax alone does not make reference data usable.

• Save order_etl.py.
• Confirm that json.JSONDecodeError is chained into ReferenceDataError with from exc.

Customer loader looks incomplete?

Keep the list check after the decoder handler. The next code chunk fills the empty customer lookup.

Need help checking the function? Help me review the first part of load_customers().

Each customer entry needs a usable identifier and name. The lookup uses the customer identifier as its key.

• Place this loop after customers = {} in load_customers().
• Keep the loop before return customers.

    for position, customer in enumerate(payload, start=1):
        if not isinstance(customer, dict):
            raise ReferenceDataError(
                "Customer item {} must be an object".format(position)
            )

        customer_id = customer.get("customer_id")
        customer_name = customer.get("name")
        if not isinstance(customer_id, str) or not customer_id.strip():
            raise ReferenceDataError(
                "Customer item {} has an invalid customer_id".format(position)
            )
        if not isinstance(customer_name, str) or not customer_name.strip():
            raise ReferenceDataError(
                "Customer item {} has an invalid name".format(position)
            )

        customers[customer_id.strip()] = customer_name.strip()

How is customer structure validated?

• The loop requires every list item to be a dictionary.
• The field checks reject blank or non-text identifiers.
• A separate field check rejects blank or non-text names.
• The final lookup maps each customer_id to a cleaned customer name.

• Save order_etl.py.
• Confirm that the customer loop is indented inside load_customers().

Customer lookup stays empty?

Place the loop between customers = {} and return customers. Returning before the loop leaves the lookup empty.

Still stuck? Help me fix the customer lookup loop.

A dedicated command-line flag gives the loaders a focused test path. It proves both lookups work before order processing begins.

• Add build_parser() below load_customers() by copying this code:

def build_parser():
    parser = argparse.ArgumentParser(
        description="Ingest e-commerce orders from CSV, JSON, and JSONL files."
    )
    parser.add_argument(
        "--input-dir",
        type=Path,
        default=Path("data"),
        help="Directory containing catalog.csv, customers.json, and orders.jsonl.",
    )
    parser.add_argument(
        "--check-reference",
        action="store_true",
        help="Load only the catalog and customer reference files.",
    )
    return parser

What does the parser add?

• The default input path points to the existing data directory.
• The --check-reference flag activates a focused reference-data check.
• The path conversion gives the loaders Path objects for child-path joining.

• Save order_etl.py.
• Confirm that the parser defaults to the existing data directory.

Check flag not recognised?

Check the spelling of --check-reference in build_parser(). Keep both leading hyphens.

Need help with the parser? Help me debug the reference-check argument.

• Add main() below build_parser() by copying this code:

def main():
    args = build_parser().parse_args()

    if args.check_reference:
        catalog = load_catalog(args.input_dir / "catalog.csv")
        customers = load_customers(args.input_dir / "customers.json")
        print(
            "Loaded {} catalog items and {} customers.".format(
                len(catalog), len(customers)
            )
        )
        return


if __name__ == "__main__":
    main()

How does check mode work?

The entry point parses the command-line arguments. Check mode loads both reference files before printing the size of each lookup.

This keeps the test focused on reference integrity. Order processing remains outside this step.

• Save order_etl.py.
• Compare your completed file with the cumulative reference below.

✔️ Awesome, I've got everything!

Your reference loaders are ready. Make sure order_etl.py is saved before running the check.

ⓧ I'd like to double check the full code

import argparse
import csv
import json
from pathlib import Path


ACCEPTED_FIELDS = [
    "order_id",
    "customer_name",
    "sku",
    "product_name",
    "quantity",
    "total_cents",
]


class ETLError(Exception):
    pass


class ReferenceDataError(ETLError):
    pass


class OrderRecordError(ETLError):
    def __init__(self, line_number, message):
        self.line_number = line_number
        super().__init__("Line {}: {}".format(line_number, message))


class OrderDecodeError(OrderRecordError):
    pass


class OrderValidationError(OrderRecordError):
    pass


class SourceEncodingError(ETLError):
    pass


def load_catalog(path):
    catalog = {}

    try:
        with path.open("r", encoding="utf-8", newline="") as handle:
            reader = csv.DictReader(handle)
            required_headers = {"sku", "product_name", "unit_price_cents"}
            actual_headers = set(reader.fieldnames or [])

            if not required_headers.issubset(actual_headers):
                missing = sorted(required_headers.difference(actual_headers))
                raise ReferenceDataError(
                    "Catalog is missing headers: {}".format(", ".join(missing))
                )

            for row in reader:
                sku = (row.get("sku") or "").strip()
                product_name = (row.get("product_name") or "").strip()

                try:
                    unit_price_cents = int(row.get("unit_price_cents"))
                except (TypeError, ValueError) as exc:
                    raise ReferenceDataError(
                        "Catalog line {} has an invalid unit_price_cents".format(
                            reader.line_num
                        )
                    ) from exc

                if not sku or not product_name or unit_price_cents < 0:
                    raise ReferenceDataError(
                        "Catalog line {} contains invalid values".format(
                            reader.line_num
                        )
                    )

                catalog[sku] = {
                    "product_name": product_name,
                    "unit_price_cents": unit_price_cents,
                }
    except csv.Error as exc:
        raise ReferenceDataError(
            "Could not parse catalog CSV: {}".format(path)
        ) from exc

    return catalog


def load_customers(path):
    try:
        with path.open("r", encoding="utf-8") as handle:
            payload = json.load(handle)
    except json.JSONDecodeError as exc:
        raise ReferenceDataError(
            "Could not parse customer JSON: {}".format(path)
        ) from exc

    if not isinstance(payload, list):
        raise ReferenceDataError("Customer JSON must contain a list")

    customers = {}
    for position, customer in enumerate(payload, start=1):
        if not isinstance(customer, dict):
            raise ReferenceDataError(
                "Customer item {} must be an object".format(position)
            )

        customer_id = customer.get("customer_id")
        customer_name = customer.get("name")
        if not isinstance(customer_id, str) or not customer_id.strip():
            raise ReferenceDataError(
                "Customer item {} has an invalid customer_id".format(position)
            )
        if not isinstance(customer_name, str) or not customer_name.strip():
            raise ReferenceDataError(
                "Customer item {} has an invalid name".format(position)
            )

        customers[customer_id.strip()] = customer_name.strip()

    return customers


def build_parser():
    parser = argparse.ArgumentParser(
        description="Ingest e-commerce orders from CSV, JSON, and JSONL files."
    )
    parser.add_argument(
        "--input-dir",
        type=Path,
        default=Path("data"),
        help="Directory containing catalog.csv, customers.json, and orders.jsonl.",
    )
    parser.add_argument(
        "--check-reference",
        action="store_true",
        help="Load only the catalog and customer reference files.",
    )
    return parser


def main():
    args = build_parser().parse_args()

    if args.check_reference:
        catalog = load_catalog(args.input_dir / "catalog.csv")
        customers = load_customers(args.input_dir / "customers.json")
        print(
            "Loaded {} catalog items and {} customers.".format(
                len(catalog), len(customers)
            )
        )
        return


if __name__ == "__main__":
    main()

What should match?

Compare the imports, exception names, loader indentation, parser options, and entry point. The completed file should load the existing reference files without changing seed_data.py.

Reference code looks different?

Check that both loader functions appear above build_parser(). Confirm that main() remains above the final entry-point condition.

Need help comparing the file? Help me find the difference in my reference loader.

Rehearse the file-format decisions

Interviewers often test whether you can justify a file-handling choice. Your implementation now gives each answer a concrete example.

• Explain why the catalog file uses newline="".
• Describe the value type returned for each csv.DictReader field.
• Compare json.load with json.loads.
• Describe how load_catalog() detects a missing unit_price_cents header.

Model interview answers

• The CSV module performs its own newline handling. Opening the file with newline="" prevents embedded-newline problems and extra carriage returns.
• A csv.DictReader maps field names to strings. Business values such as prices still need explicit conversion and validation.
• Use json.load for one JSON document in a file-like object. Use json.loads for JSON already held in a string or bytes object.
• Create a required-header set. Compare it with reader.fieldnames. Raise ReferenceDataError with the missing names before reading rows.

Before you run the final check, what catalog count and customer count do you expect from the seeded files?

• Verify both reference lookups by running this command in the integrated terminal:

python3 order_etl.py --check-reference

What does this command do?

The command runs order_etl.py with the reference-only flag. The script loads data/catalog.csv before loading data/customers.json.

You should see Loaded 2 catalog items and 2 customers. in the terminal. This proves both UTF-8 reference files passed parsing and structural validation.

That reference boundary is working. The pipeline can now identify products and customers without rereading either file for every order.

Reference check not completing?

Confirm that the integrated terminal is still inside the ecommerce-order-etl folder. The script expects the data directory beside order_etl.py.

Check the first row of data/catalog.csv for the three required headers. Check that data/customers.json still contains one complete JSON list.

Still stuck? Help me debug my reference-data check.

Your catalog and customer lookups are ready. Next, you will stream the order feed one record at a time and expose its malformed second line.

Stream Orders in Strict Mode

Your reference loaders can now turn the product catalog and customer document into keyed lookups. The growing order feed still needs a safe path through the pipeline.

Loading the entire feed would hide its memory cost and blur the failure boundary. This step uses JSONL streaming to process one order at a time before strict mode exposes the malformed second record.

In this step, get ready to:

• Stream nonblank order records with their source line numbers.
• Validate each order against the catalog and customer lookups.
• Run strict mode to inspect a chained record failure and its partial outputs.

Stream and parse the JSONL feed

Each nonblank line in a JSONL file represents one independent record. Direct iteration keeps the current line in memory while preserving its location in the source.

• Switch back to order_etl.py in the VS Code editor.
• Place the line-streaming function below load_customers by pasting this code:

def iter_jsonl_lines(path):
    with path.open("r", encoding="utf-8") as handle:
        for line_number, raw_line in enumerate(handle, start=1):
            stripped_line = raw_line.strip()
            if stripped_line:
                yield line_number, stripped_line

What does the line iterator do?

• The context manager closes the source file if iteration finishes or raises an exception.
• The explicit utf-8 encoding gives the source a clear decoding contract.
• The generator yields one-based line numbers so every rejected order can point back to its source.

• Save order_etl.py.
• Confirm the existing reference path still runs by executing this command:

python3 order_etl.py --check-reference

What does this check prove?

You should still see Loaded 2 catalog items and 2 customers. The unchanged result confirms the new generator did not break the runnable reference path.

Reference check stopped?

Check that iter_jsonl_lines begins at the left edge of the file. Keep its remaining lines indented beneath the function definition.

If the terminal points to a syntax or indentation problem, help me compare my iter_jsonl_lines function with the expected structure.

The iterator preserves raw record boundaries. The parser can now attach a domain-specific error to a single malformed line through exception chaining.

• Add the record parser directly below iter_jsonl_lines by pasting this code:

def parse_order_line(raw_line, line_number):
    try:
        order = json.loads(raw_line)
    except json.JSONDecodeError as exc:
        raise OrderDecodeError(
            line_number,
            "Invalid JSON at column {}: {}".format(exc.colno, exc.msg),
        ) from exc

    if not isinstance(order, dict):
        raise OrderValidationError(
            line_number, "Each JSONL line must contain a JSON object"
        )

    return order

How does the parser preserve context?

• The json.loads call parses one line because the iterator already separated the feed into records.
• The OrderDecodeError message adds the source line and JSON column for an operator.
• The raise ... from ... chain retains the original decoder details for a developer.

• Save order_etl.py.
• Check the updated script for syntax problems by running:

python3 order_etl.py --check-reference

What should the script report?

You should see the same two catalog items and two customers. The result proves Python can define both streaming functions before entering the existing reference-check path.

Validate and enrich each order

Successful JSON decoding only proves that a line follows JSON syntax. Business validation must still reject missing fields, invalid quantities, unknown customers, or unknown products.

• Add the text-field helper below parse_order_line by pasting this code:

def require_text(order, field, line_number):
    value = order[field]
    if not isinstance(value, str) or not value.strip():
        raise OrderValidationError(
            line_number, "{} must be a non-empty string".format(field)
        )
    return value.strip()

Why centralize text validation?

The helper applies one rule to every required text field. It also keeps the source line number attached to each validation failure.

• Save order_etl.py.
• Confirm the helper leaves the reference check runnable by executing:

python3 order_etl.py --check-reference

What does the repeated result confirm?

You should still see the successful reference count. This run confirms the helper is defined correctly before the order transformation starts using it.

The transformation combines schema checks with business lookups. It keeps prices in integer cents so every order total uses exact whole-number arithmetic.

• Start transform_order directly below require_text by pasting this first section:

def transform_order(order, line_number, catalog, customers):
    required_fields = ("order_id", "customer_id", "sku", "quantity")
    missing_fields = [field for field in required_fields if field not in order]
    if missing_fields:
        raise OrderValidationError(
            line_number,
            "Missing fields: {}".format(", ".join(missing_fields)),
        )

    order_id = require_text(order, "order_id", line_number)
    customer_id = require_text(order, "customer_id", line_number)
    sku = require_text(order, "sku", line_number)
    raw_quantity = order["quantity"]

    if isinstance(raw_quantity, bool):
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )
    if isinstance(raw_quantity, int):
        quantity = raw_quantity
    elif isinstance(raw_quantity, str) and raw_quantity.strip().isdigit():
        quantity = int(raw_quantity.strip())
    else:
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )

What does the first transformation section check?

• The required-field check reports every missing order key in one domain error.
• The text helper normalizes identifiers before the lookup stage.
• The quantity rules accept positive integers or digit-only strings while rejecting booleans and other types.

• Complete transform_order by pasting this section immediately below the quantity validation:

    if quantity <= 0:
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )
    if customer_id not in customers:
        raise OrderValidationError(
            line_number, "Unknown customer_id: {}".format(customer_id)
        )
    if sku not in catalog:
        raise OrderValidationError(
            line_number, "Unknown sku: {}".format(sku)
        )

    product = catalog[sku]
    return {
        "order_id": order_id,
        "customer_name": customers[customer_id],
        "sku": sku,
        "product_name": product["product_name"],
        "quantity": quantity,
        "total_cents": product["unit_price_cents"] * quantity,
    }

How does enrichment work?

The lookup checks prevent unknown customers or products from entering the accepted output. A valid order receives the customer name, product name, normalized quantity, and calculated total.

• Save order_etl.py.
• Verify the completed transformation has valid syntax by running:

python3 order_etl.py --check-reference

What should remain unchanged?

You should still see two catalog items and two customers. The reference result now doubles as a quick syntax check for the complete transformation function.

Reference check failing after transformation?

Confirm the second transformation section stays indented inside transform_order. Check that the returned dictionary closes before the next function begins.

Need another set of eyes? Help me find the syntax or indentation problem in my transform_order function.

Run strict mode and rehearse the failure

Strict mode applies a fail-fast policy to record errors. The run still preserves the rejected line and the partial accepted output before re-raising the domain exception.

• Add the quarantine writer below transform_order by pasting this code:

def write_rejection(handle, line_number, raw_line, exc):
    rejection = {
        "line_number": line_number,
        "error_type": type(exc).__name__,
        "message": str(exc),
        "raw_record": raw_line,
    }
    handle.write(json.dumps(rejection, ensure_ascii=False) + "\n")

What does the rejection preserve?

Each rejection records the source line, domain error type, readable message, and original record. JSONL keeps every rejected record on its own line for later diagnosis.

• Save order_etl.py.
• Confirm the script still loads by running:

python3 order_etl.py --check-reference

What does this successful run establish?

The reference count should appear again. This confirms the rejection writer is available without changing the existing command path.

The execution function coordinates the reference lookups, streaming loop, output files, and run summary. Paste the following four sections in order to keep the single function aligned.

• Start run_etl below write_rejection by pasting this first section:

def run_etl(input_dir, output_dir, strict=False):
    summary = {"accepted": 0, "rejected": 0, "status": "running"}
    summary_path = output_dir / "run_summary.json"

    try:
        output_dir.mkdir(parents=True, exist_ok=True)
        catalog = load_catalog(input_dir / "catalog.csv")
        customers = load_customers(input_dir / "customers.json")

        accepted_path = output_dir / "accepted_orders.csv"
        rejected_path = output_dir / "rejected_orders.jsonl"

        with accepted_path.open(
            "w", encoding="utf-8", newline=""
        ) as accepted_handle, rejected_path.open(
            "w", encoding="utf-8"
        ) as rejected_handle:
            accepted_writer = csv.DictWriter(
                accepted_handle, fieldnames=ACCEPTED_FIELDS
            )
            accepted_writer.writeheader()

What does the run setup control?

The summary begins in a running state before the lookups load. One outer context manager owns both output handles so strict failure closes both files cleanly.

• Continue the same run_etl function by pasting this processing loop directly below accepted_writer.writeheader():

            for line_number, raw_line in iter_jsonl_lines(
                input_dir / "orders.jsonl"
            ):
                try:
                    order = parse_order_line(raw_line, line_number)
                    accepted_order = transform_order(
                        order, line_number, catalog, customers
                    )
                except OrderRecordError as exc:
                    summary["rejected"] += 1
                    write_rejection(
                        rejected_handle, line_number, raw_line, exc
                    )
                    if strict:
                        raise
                    continue

                accepted_writer.writerow(accepted_order)
                summary["accepted"] += 1

        summary["status"] = "completed"

Where is the strict boundary?

Expected record errors are caught inside the loop so the raw record can be quarantined. The strict check re-raises the same domain exception immediately after the rejection is written.

• Continue run_etl by pasting these run-level exception handlers below the completed-status line:

    except UnicodeDecodeError as exc:
        summary["status"] = "failed"
        raise SourceEncodingError(
            "Could not decode an input file as UTF-8 near byte offset {}".format(
                exc.start
            )
        ) from exc
    except (OSError, csv.Error) as exc:
        summary["status"] = "failed"
        raise ETLError("File operation failed: {}".format(exc)) from exc
    except ETLError:
        summary["status"] = "failed"
        raise
    except Exception:
        summary["status"] = "failed"
        raise

Why separate the exception boundaries?

Source decoding and file failures describe an untrustworthy stream. Domain ETL errors keep their existing category while unexpected programming errors are marked as failed and re-raised.

• Finish run_etl by pasting this final section directly below the last exception handler:

    finally:
        try:
            output_dir.mkdir(parents=True, exist_ok=True)
            summary_path.write_text(
                json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
        except OSError as summary_error:
            print("Could not write run summary: {}".format(summary_error))

    return summary

Why write the summary in finally?

The finally block attempts the summary write after success or failure. The status and counters therefore describe how far the run progressed before control returned or an exception escaped.

• Save order_etl.py.
• Check the complete execution function for syntax problems by running:

python3 order_etl.py --check-reference

What should this checkpoint show?

You should see the familiar reference count. That result confirms Python can define the complete ETL function before strict mode calls it.

Script stopped after adding run_etl?

Check that every except and the finally align with the outer try. Keep the processing loop nested inside the two-file context manager.

If the nesting is difficult to spot, help me diagnose the indentation in my run_etl function.

• Add the strict option inside build_parser immediately above the existing --check-reference option:

    parser.add_argument(
        "--strict",
        action="store_true",
        help="Stop after the first invalid order.",
    )

What does the strict option control?

The parser stores whether strict mode was requested. That boolean reaches the record loop through the run_etl call.

• Add the normal execution path inside main immediately after the reference-check branch returns:

    summary = run_etl(args.input_dir, args.output_dir, strict=args.strict)
    print(
        "ETL succeeded: {} accepted, {} rejected.".format(
            summary["accepted"], summary["rejected"]
        )
    )

How does main start the pipeline?

The parsed input directory, output directory, and strict flag are passed into run_etl. The success message only runs when no exception escapes.

• Save order_etl.py.

✔️ Awesome, I've got everything!

Your strict streaming path is ready. Keep order_etl.py saved before running the designed obstacle.

ⓧ I'd like to double check the full code

import argparse
import csv
import json
from pathlib import Path


ACCEPTED_FIELDS = [
    "order_id",
    "customer_name",
    "sku",
    "product_name",
    "quantity",
    "total_cents",
]


class ETLError(Exception):
    pass


class ReferenceDataError(ETLError):
    pass


class OrderRecordError(ETLError):
    def __init__(self, line_number, message):
        self.line_number = line_number
        super().__init__("Line {}: {}".format(line_number, message))


class OrderDecodeError(OrderRecordError):
    pass


class OrderValidationError(OrderRecordError):
    pass


class SourceEncodingError(ETLError):
    pass


def load_catalog(path):
    catalog = {}

    try:
        with path.open("r", encoding="utf-8", newline="") as handle:
            reader = csv.DictReader(handle)
            required_headers = {"sku", "product_name", "unit_price_cents"}
            actual_headers = set(reader.fieldnames or [])

            if not required_headers.issubset(actual_headers):
                missing = sorted(required_headers.difference(actual_headers))
                raise ReferenceDataError(
                    "Catalog is missing headers: {}".format(", ".join(missing))
                )

            for row in reader:
                sku = (row.get("sku") or "").strip()
                product_name = (row.get("product_name") or "").strip()

                try:
                    unit_price_cents = int(row.get("unit_price_cents"))
                except (TypeError, ValueError) as exc:
                    raise ReferenceDataError(
                        "Catalog line {} has an invalid unit_price_cents".format(
                            reader.line_num
                        )
                    ) from exc

                if not sku or not product_name or unit_price_cents < 0:
                    raise ReferenceDataError(
                        "Catalog line {} contains invalid values".format(
                            reader.line_num
                        )
                    )

                catalog[sku] = {
                    "product_name": product_name,
                    "unit_price_cents": unit_price_cents,
                }
    except csv.Error as exc:
        raise ReferenceDataError(
            "Could not parse catalog CSV: {}".format(path)
        ) from exc

    return catalog


def load_customers(path):
    try:
        with path.open("r", encoding="utf-8") as handle:
            payload = json.load(handle)
    except json.JSONDecodeError as exc:
        raise ReferenceDataError(
            "Could not parse customer JSON: {}".format(path)
        ) from exc

    if not isinstance(payload, list):
        raise ReferenceDataError("Customer JSON must contain a list")

    customers = {}
    for position, customer in enumerate(payload, start=1):
        if not isinstance(customer, dict):
            raise ReferenceDataError(
                "Customer item {} must be an object".format(position)
            )

        customer_id = customer.get("customer_id")
        customer_name = customer.get("name")
        if not isinstance(customer_id, str) or not customer_id.strip():
            raise ReferenceDataError(
                "Customer item {} has an invalid customer_id".format(position)
            )
        if not isinstance(customer_name, str) or not customer_name.strip():
            raise ReferenceDataError(
                "Customer item {} has an invalid name".format(position)
            )

        customers[customer_id.strip()] = customer_name.strip()

    return customers


def iter_jsonl_lines(path):
    with path.open("r", encoding="utf-8") as handle:
        for line_number, raw_line in enumerate(handle, start=1):
            stripped_line = raw_line.strip()
            if stripped_line:
                yield line_number, stripped_line


def parse_order_line(raw_line, line_number):
    try:
        order = json.loads(raw_line)
    except json.JSONDecodeError as exc:
        raise OrderDecodeError(
            line_number,
            "Invalid JSON at column {}: {}".format(exc.colno, exc.msg),
        ) from exc

    if not isinstance(order, dict):
        raise OrderValidationError(
            line_number, "Each JSONL line must contain a JSON object"
        )

    return order


def require_text(order, field, line_number):
    value = order[field]
    if not isinstance(value, str) or not value.strip():
        raise OrderValidationError(
            line_number, "{} must be a non-empty string".format(field)
        )
    return value.strip()


def transform_order(order, line_number, catalog, customers):
    required_fields = ("order_id", "customer_id", "sku", "quantity")
    missing_fields = [field for field in required_fields if field not in order]
    if missing_fields:
        raise OrderValidationError(
            line_number,
            "Missing fields: {}".format(", ".join(missing_fields)),
        )

    order_id = require_text(order, "order_id", line_number)
    customer_id = require_text(order, "customer_id", line_number)
    sku = require_text(order, "sku", line_number)
    raw_quantity = order["quantity"]

    if isinstance(raw_quantity, bool):
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )
    if isinstance(raw_quantity, int):
        quantity = raw_quantity
    elif isinstance(raw_quantity, str) and raw_quantity.strip().isdigit():
        quantity = int(raw_quantity.strip())
    else:
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )

    if quantity <= 0:
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )
    if customer_id not in customers:
        raise OrderValidationError(
            line_number, "Unknown customer_id: {}".format(customer_id)
        )
    if sku not in catalog:
        raise OrderValidationError(
            line_number, "Unknown sku: {}".format(sku)
        )

    product = catalog[sku]
    return {
        "order_id": order_id,
        "customer_name": customers[customer_id],
        "sku": sku,
        "product_name": product["product_name"],
        "quantity": quantity,
        "total_cents": product["unit_price_cents"] * quantity,
    }


def write_rejection(handle, line_number, raw_line, exc):
    rejection = {
        "line_number": line_number,
        "error_type": type(exc).__name__,
        "message": str(exc),
        "raw_record": raw_line,
    }
    handle.write(json.dumps(rejection, ensure_ascii=False) + "\n")


def run_etl(input_dir, output_dir, strict=False):
    summary = {"accepted": 0, "rejected": 0, "status": "running"}
    summary_path = output_dir / "run_summary.json"

    try:
        output_dir.mkdir(parents=True, exist_ok=True)
        catalog = load_catalog(input_dir / "catalog.csv")
        customers = load_customers(input_dir / "customers.json")

        accepted_path = output_dir / "accepted_orders.csv"
        rejected_path = output_dir / "rejected_orders.jsonl"

        with accepted_path.open(
            "w", encoding="utf-8", newline=""
        ) as accepted_handle, rejected_path.open(
            "w", encoding="utf-8"
        ) as rejected_handle:
            accepted_writer = csv.DictWriter(
                accepted_handle, fieldnames=ACCEPTED_FIELDS
            )
            accepted_writer.writeheader()

            for line_number, raw_line in iter_jsonl_lines(
                input_dir / "orders.jsonl"
            ):
                try:
                    order = parse_order_line(raw_line, line_number)
                    accepted_order = transform_order(
                        order, line_number, catalog, customers
                    )
                except OrderRecordError as exc:
                    summary["rejected"] += 1
                    write_rejection(
                        rejected_handle, line_number, raw_line, exc
                    )
                    if strict:
                        raise
                    continue

                accepted_writer.writerow(accepted_order)
                summary["accepted"] += 1

        summary["status"] = "completed"
    except UnicodeDecodeError as exc:
        summary["status"] = "failed"
        raise SourceEncodingError(
            "Could not decode an input file as UTF-8 near byte offset {}".format(
                exc.start
            )
        ) from exc
    except (OSError, csv.Error) as exc:
        summary["status"] = "failed"
        raise ETLError("File operation failed: {}".format(exc)) from exc
    except ETLError:
        summary["status"] = "failed"
        raise
    except Exception:
        summary["status"] = "failed"
        raise
    finally:
        try:
            output_dir.mkdir(parents=True, exist_ok=True)
            summary_path.write_text(
                json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
        except OSError as summary_error:
            print("Could not write run summary: {}".format(summary_error))

    return summary


def build_parser():
    parser = argparse.ArgumentParser(
        description="Ingest e-commerce orders from CSV, JSON, and JSONL files."
    )
    parser.add_argument(
        "--input-dir",
        type=Path,
        default=Path("data"),
        help="Directory containing catalog.csv, customers.json, and orders.jsonl.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("output"),
        help="Directory for accepted, rejected, and summary files.",
    )
    parser.add_argument(
        "--strict",
        action="store_true",
        help="Stop after the first invalid order.",
    )
    parser.add_argument(
        "--check-reference",
        action="store_true",
        help="Load only the catalog and customer reference files.",
    )
    return parser


def main():
    args = build_parser().parse_args()

    if args.check_reference:
        catalog = load_catalog(args.input_dir / "catalog.csv")
        customers = load_customers(args.input_dir / "customers.json")
        print(
            "Loaded {} catalog items and {} customers.".format(
                len(catalog), len(customers)
            )
        )
        return

    summary = run_etl(args.input_dir, args.output_dir, strict=args.strict)
    print(
        "ETL succeeded: {} accepted, {} rejected.".format(
            summary["accepted"], summary["rejected"]
        )
    )


if __name__ == "__main__":
    main()

How to use this reference

Compare each function and indentation level with your saved order_etl.py. The reference includes the cumulative file needed for the strict run.

Before you run strict mode, where do you think processing stops when the malformed second line reaches the parser?

• Run the strict ETL path from the existing VS Code terminal with this command:

python3 order_etl.py --strict

This failure is intentional

The traceback ends with OrderDecodeError for line 2. The earlier traceback section shows the original json.JSONDecodeError as its direct cause.

Strict mode accepted the valid first order before the malformed record stopped the run. This is the failure boundary you set out to expose.

• Expand the output folder in the VS Code Explorer.
• Open accepted_orders.csv to inspect the partial accepted output.
• Open rejected_orders.jsonl to inspect the quarantined malformed record.
• Open run_summary.json to inspect the failed run counters.

You should see the header and order O-1001 in accepted_orders.csv. The rejected file should contain one OrderDecodeError entry for source line 2.

The summary should report one accepted record, one rejected record, and a failed status. That partial state proves the finally block ran after the exception escaped.

Strict run did not stop on line 2?

Confirm data/orders.jsonl still contains the malformed second line created by seed_data.py. Check that if strict: remains inside the OrderRecordError handler.

If the output counts differ, help me trace why my strict ETL did not stop after rejecting line 2.

The visible failure gives you a concrete interview story about memory use and recovery boundaries. Rehearse each answer from the behavior you just observed.

Interview Drill: Large Files and JSONL

• Question: Why is direct file iteration safer than read() for a very large source? Model answer: read() without a size returns the entire file. Direct iteration keeps only the current record in the per-order working set.
• Question: Why does JSONL suit record-level recovery better than one large JSON array? Model answer: Each line is an independent JSON value. The pipeline can attach a line number to one bad record without reparsing the full document.
• Question: Where should parsing exceptions be caught in a streaming pipeline? Model answer: Catch expected record errors inside the loop when processing can continue. Catch source-level failures outside the loop because the stream itself is no longer trustworthy.
• Coding problem: Convert a whole-file JSON approach into a stream of numbered raw lines. Model approach: Use enumerate(handle, start=1) inside a generator. Skip blank lines before yielding each pair.

You now have a strict streaming pipeline that preserves its first accepted order and exposes the malformed record through a chained custom exception. Next, you will turn that visible failure into a resilient quarantine policy.

Quarantine Bad Orders and Finish the ETL

Your strict run proved that a malformed order reaches a domain-specific exception. It also left partial output files plus a failed run summary.

A production-minded ETL needs an explicit policy for bad records. This step quarantines expected record defects while keeping valid orders moving.

In this step, get ready to:

• Continue processing after expected record-level failures.
• Convert source failures at the outer ETL boundary.
• Defend the finished failure policy in interview answers.

Recover from record-level errors

Each line in the JSONL feed represents an independent order. Catching OrderRecordError inside the loop lets the ETL preserve one bad line without losing later valid orders.

• Switch back to order_etl.py in your VS Code workspace.
• In run_etl(), find the block beginning with except OrderRecordError as exc:.
• Replace that exception block with the code below:

                except OrderRecordError as exc:
                    summary["rejected"] += 1
                    write_rejection(
                        rejected_handle, line_number, raw_line, exc
                    )
                    if strict:
                        raise
                    continue

What Does This Recovery Block Do?

• The OrderRecordError handler catches expected decoding or validation defects for one order.
• The write_rejection() call preserves the source line plus diagnostic context.
• The if strict: branch preserves fail-fast behavior. The final continue advances resilient runs to the next record.

• Save order_etl.py.

Before you run resilient mode, how many orders do you expect the pipeline to accept after it moves malformed or invalid records into quarantine?

• Run the ETL without strict mode by running this command:

python3 order_etl.py

What Should Happen?

The terminal prints ETL succeeded: 2 accepted, 2 rejected. The pipeline accepts lines 1 and 4.

The malformed JSON on line 2 becomes one rejection. The unknown SKU on line 3 becomes the other rejection.

• Select output/accepted_orders.csv in the VS Code Explorer.
• Confirm the accepted file contains orders O-1001 and O-1004.
• Select output/rejected_orders.jsonl in the VS Code Explorer.
• Confirm the rejection file contains one OrderDecodeError plus one OrderValidationError.

Your valid orders now reach the accepted CSV even when another record is malformed. The rejected lines remain available for diagnosis.

Still Seeing a Failed Run?

Check that raise sits inside the if strict: block. A raise at the same indentation as continue stops every run.

Confirm your command does not include the strict flag. Regenerate the source files with the seeder from Step 1 if their contents have changed.

Still stuck? Help me debug why non-strict order processing still stops at the malformed JSONL record.

Separate source failures and guarantee the summary

Your strict path already writes its summary from finally. The surrounding handlers now need to distinguish a broken source from an isolated bad record.

A UnicodeDecodeError means Python could not produce trustworthy text from the source bytes. That failure belongs outside the record loop because the stream itself can no longer be trusted.

• In run_etl(), locate the except ETLError: handler after the order-processing block.
• Replace that handler plus the existing finally tail with the code below:

    except UnicodeDecodeError as exc:
        summary["status"] = "failed"
        raise SourceEncodingError(
            "Could not decode an input file as UTF-8 near byte offset {}".format(
                exc.start
            )
        ) from exc
    except (OSError, csv.Error) as exc:
        summary["status"] = "failed"
        raise ETLError("File operation failed: {}".format(exc)) from exc
    except ETLError:
        summary["status"] = "failed"
        raise
    except Exception:
        summary["status"] = "failed"
        raise
    finally:
        try:
            output_dir.mkdir(parents=True, exist_ok=True)
            summary_path.write_text(
                json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
        except OSError as summary_error:
            print("Could not write run summary: {}".format(summary_error))

How Does the Source Boundary Work?

• The first handler chains a decoding failure into SourceEncodingError with the failing byte offset.
• The next handler wraps filesystem or CSV failures as ETLError.
• The broad final handler marks unexpected programming failures without misclassifying them as bad order data.
• The finally block attempts the run summary after successful processing or failure.

• Save order_etl.py.

Before you run strict mode again, do you expect its output artifacts to be complete or partial?

• Re-run the expected strict-mode failure by running this command:

python3 order_etl.py --strict

This Failure Is Expected

The first order is accepted before malformed line 2 raises OrderDecodeError. The exception escapes because strict mode preserves fail-fast behavior.

The finally block still writes a failed summary. This proves cleanup reporting survives the error path.

• Switch to output/accepted_orders.csv.
• Confirm it contains only the header plus order O-1001.
• Switch to output/rejected_orders.jsonl.
• Confirm it contains only the rejection for line 2.
• Switch to output/run_summary.json.
• Confirm it records one accepted order plus one rejected order with a failed status.

Strict mode leaves both order-output files partial. The summary makes that partial state visible to operators.

Summary Missing After the Failure?

Check that finally: aligns with the outer exception handlers. If it sits inside one handler, other failure paths can bypass it.

Check that summary_path still points to the output_dir path created near the start of run_etl().

Need a second pair of eyes? Help me trace why run_summary.json is not written after my strict ETL fails.

Your record and source boundaries should now match the finished implementation. Use the tabs below to compare the complete file.

✔️ Awesome, I've got everything!

Great work. Save order_etl.py before moving to the interview drill.

ⓧ I'd like to double check the full code

import argparse
import csv
import json
from pathlib import Path


ACCEPTED_FIELDS = [
    "order_id",
    "customer_name",
    "sku",
    "product_name",
    "quantity",
    "total_cents",
]


class ETLError(Exception):
    pass


class ReferenceDataError(ETLError):
    pass


class OrderRecordError(ETLError):
    def __init__(self, line_number, message):
        self.line_number = line_number
        super().__init__("Line {}: {}".format(line_number, message))


class OrderDecodeError(OrderRecordError):
    pass


class OrderValidationError(OrderRecordError):
    pass


class SourceEncodingError(ETLError):
    pass


def load_catalog(path):
    catalog = {}

    try:
        with path.open("r", encoding="utf-8", newline="") as handle:
            reader = csv.DictReader(handle)
            required_headers = {"sku", "product_name", "unit_price_cents"}
            actual_headers = set(reader.fieldnames or [])

            if not required_headers.issubset(actual_headers):
                missing = sorted(required_headers.difference(actual_headers))
                raise ReferenceDataError(
                    "Catalog is missing headers: {}".format(", ".join(missing))
                )

            for row in reader:
                sku = (row.get("sku") or "").strip()
                product_name = (row.get("product_name") or "").strip()

                try:
                    unit_price_cents = int(row.get("unit_price_cents"))
                except (TypeError, ValueError) as exc:
                    raise ReferenceDataError(
                        "Catalog line {} has an invalid unit_price_cents".format(
                            reader.line_num
                        )
                    ) from exc

                if not sku or not product_name or unit_price_cents < 0:
                    raise ReferenceDataError(
                        "Catalog line {} contains invalid values".format(
                            reader.line_num
                        )
                    )

                catalog[sku] = {
                    "product_name": product_name,
                    "unit_price_cents": unit_price_cents,
                }
    except csv.Error as exc:
        raise ReferenceDataError(
            "Could not parse catalog CSV: {}".format(path)
        ) from exc

    return catalog


def load_customers(path):
    try:
        with path.open("r", encoding="utf-8") as handle:
            payload = json.load(handle)
    except json.JSONDecodeError as exc:
        raise ReferenceDataError(
            "Could not parse customer JSON: {}".format(path)
        ) from exc

    if not isinstance(payload, list):
        raise ReferenceDataError("Customer JSON must contain a list")

    customers = {}
    for position, customer in enumerate(payload, start=1):
        if not isinstance(customer, dict):
            raise ReferenceDataError(
                "Customer item {} must be an object".format(position)
            )

        customer_id = customer.get("customer_id")
        customer_name = customer.get("name")
        if not isinstance(customer_id, str) or not customer_id.strip():
            raise ReferenceDataError(
                "Customer item {} has an invalid customer_id".format(position)
            )
        if not isinstance(customer_name, str) or not customer_name.strip():
            raise ReferenceDataError(
                "Customer item {} has an invalid name".format(position)
            )

        customers[customer_id.strip()] = customer_name.strip()

    return customers


def iter_jsonl_lines(path):
    with path.open("r", encoding="utf-8") as handle:
        for line_number, raw_line in enumerate(handle, start=1):
            stripped_line = raw_line.strip()
            if stripped_line:
                yield line_number, stripped_line


def parse_order_line(raw_line, line_number):
    try:
        order = json.loads(raw_line)
    except json.JSONDecodeError as exc:
        raise OrderDecodeError(
            line_number,
            "Invalid JSON at column {}: {}".format(exc.colno, exc.msg),
        ) from exc

    if not isinstance(order, dict):
        raise OrderValidationError(
            line_number, "Each JSONL line must contain a JSON object"
        )

    return order


def require_text(order, field, line_number):
    value = order[field]
    if not isinstance(value, str) or not value.strip():
        raise OrderValidationError(
            line_number, "{} must be a non-empty string".format(field)
        )
    return value.strip()


def transform_order(order, line_number, catalog, customers):
    required_fields = ("order_id", "customer_id", "sku", "quantity")
    missing_fields = [field for field in required_fields if field not in order]
    if missing_fields:
        raise OrderValidationError(
            line_number,
            "Missing fields: {}".format(", ".join(missing_fields)),
        )

    order_id = require_text(order, "order_id", line_number)
    customer_id = require_text(order, "customer_id", line_number)
    sku = require_text(order, "sku", line_number)
    raw_quantity = order["quantity"]

    if isinstance(raw_quantity, bool):
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )
    if isinstance(raw_quantity, int):
        quantity = raw_quantity
    elif isinstance(raw_quantity, str) and raw_quantity.strip().isdigit():
        quantity = int(raw_quantity.strip())
    else:
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )

    if quantity <= 0:
        raise OrderValidationError(
            line_number, "quantity must be a positive integer"
        )
    if customer_id not in customers:
        raise OrderValidationError(
            line_number, "Unknown customer_id: {}".format(customer_id)
        )
    if sku not in catalog:
        raise OrderValidationError(
            line_number, "Unknown sku: {}".format(sku)
        )

    product = catalog[sku]
    return {
        "order_id": order_id,
        "customer_name": customers[customer_id],
        "sku": sku,
        "product_name": product["product_name"],
        "quantity": quantity,
        "total_cents": product["unit_price_cents"] * quantity,
    }


def write_rejection(handle, line_number, raw_line, exc):
    rejection = {
        "line_number": line_number,
        "error_type": type(exc).__name__,
        "message": str(exc),
        "raw_record": raw_line,
    }
    handle.write(json.dumps(rejection, ensure_ascii=False) + "\n")


def run_etl(input_dir, output_dir, strict=False):
    summary = {"accepted": 0, "rejected": 0, "status": "running"}
    summary_path = output_dir / "run_summary.json"

    try:
        output_dir.mkdir(parents=True, exist_ok=True)
        catalog = load_catalog(input_dir / "catalog.csv")
        customers = load_customers(input_dir / "customers.json")

        accepted_path = output_dir / "accepted_orders.csv"
        rejected_path = output_dir / "rejected_orders.jsonl"

        with accepted_path.open(
            "w", encoding="utf-8", newline=""
        ) as accepted_handle, rejected_path.open(
            "w", encoding="utf-8"
        ) as rejected_handle:
            accepted_writer = csv.DictWriter(
                accepted_handle, fieldnames=ACCEPTED_FIELDS
            )
            accepted_writer.writeheader()

            for line_number, raw_line in iter_jsonl_lines(
                input_dir / "orders.jsonl"
            ):
                try:
                    order = parse_order_line(raw_line, line_number)
                    accepted_order = transform_order(
                        order, line_number, catalog, customers
                    )
                except OrderRecordError as exc:
                    summary["rejected"] += 1
                    write_rejection(
                        rejected_handle, line_number, raw_line, exc
                    )
                    if strict:
                        raise
                    continue

                accepted_writer.writerow(accepted_order)
                summary["accepted"] += 1

        summary["status"] = "completed"
    except UnicodeDecodeError as exc:
        summary["status"] = "failed"
        raise SourceEncodingError(
            "Could not decode an input file as UTF-8 near byte offset {}".format(
                exc.start
            )
        ) from exc
    except (OSError, csv.Error) as exc:
        summary["status"] = "failed"
        raise ETLError("File operation failed: {}".format(exc)) from exc
    except ETLError:
        summary["status"] = "failed"
        raise
    except Exception:
        summary["status"] = "failed"
        raise
    finally:
        try:
            output_dir.mkdir(parents=True, exist_ok=True)
            summary_path.write_text(
                json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
        except OSError as summary_error:
            print("Could not write run summary: {}".format(summary_error))

    return summary


def build_parser():
    parser = argparse.ArgumentParser(
        description="Ingest e-commerce orders from CSV, JSON, and JSONL files."
    )
    parser.add_argument(
        "--input-dir",
        type=Path,
        default=Path("data"),
        help="Directory containing catalog.csv, customers.json, and orders.jsonl.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("output"),
        help="Directory for accepted, rejected, and summary files.",
    )
    parser.add_argument(
        "--strict",
        action="store_true",
        help="Stop after the first invalid order.",
    )
    parser.add_argument(
        "--check-reference",
        action="store_true",
        help="Load only the catalog and customer reference files.",
    )
    return parser


def main():
    args = build_parser().parse_args()

    if args.check_reference:
        catalog = load_catalog(args.input_dir / "catalog.csv")
        customers = load_customers(args.input_dir / "customers.json")
        print(
            "Loaded {} catalog items and {} customers.".format(
                len(catalog), len(customers)
            )
        )
        return

    summary = run_etl(args.input_dir, args.output_dir, strict=args.strict)
    print(
        "ETL succeeded: {} accepted, {} rejected.".format(
            summary["accepted"], summary["rejected"]
        )
    )


if __name__ == "__main__":
    main()

Rehearse the production trade-offs

The finished pipeline gives you concrete examples for interview questions about error boundaries and resource cleanup. Try each answer aloud before checking the model response.

✔️ I'll try it first

• Why should you avoid catching bare Exception around every row? Answer aloud before checking the model response.
• What information belongs in a custom ETL exception? Answer aloud before checking the model response.
• When should you use finally instead of with? Answer aloud before checking the model response.
• Which failures should stop the source? Which failures can enter quarantine? Answer both parts aloud.
• What are the time and memory costs of this pipeline? Answer aloud before checking the model response.
• How would you detect duplicate order_id values? Include the memory trade-off in your answer.

ⓧ Let me check my answers

• Bare row-level catches can misclassify programmer defects as bad data. Catch expected domain exceptions locally. Mark unexpected exceptions as failed at the run boundary before re-raising them.
• A custom ETL exception needs a stable domain category. It also needs actionable context such as the source or line number. Field details plus the original cause help developers diagnose the failure.
• Use with for files or other context-managed resources. Use finally for run-level actions that must be attempted after success or failure.
• Fail fast when reference data is corrupt or the source cannot be trusted. Quarantine isolated record defects when partial success is allowed and rejected data remains visible.
• Processing takes O(n) time for n orders. Each streamed order uses a constant working set. The customer and product lookups require O(c + p) memory.
• Maintain a set of previously seen order_id values. Raise OrderValidationError when a value repeats. An unbounded feed may require an external state store because the set grows with every unique order.

Before the final run, predict which status replaces the failed status left by strict mode.

• Restore the finished resilient artifacts by running this command:

python3 order_etl.py

What Proves the ETL Is Complete?

The terminal reports ETL succeeded: 2 accepted, 2 rejected. This run replaces the strict-mode partial artifacts.

The summary now records two accepted orders plus two rejected orders with a completed status.

• Confirm output/accepted_orders.csv contains O-1001 plus O-1004.
• Confirm output/rejected_orders.jsonl contains rejections for lines 2 and 3.
• Confirm output/run_summary.json records "status": "completed".

You have closed the main failure gap. Your ETL now preserves valid business data while keeping every rejected record observable.

Secret mission

Break the Feed's Encoding

Your ETL can quarantine malformed records. Now corrupt the order feed at the byte level. Then prove why decoding failures must stop the source.

Clean Up Your Resources

Clean Up Your Resources

Your ETL runs entirely on local files with the Python standard library, so there are no ongoing costs. Choose whether to keep the lab ready, pause your work, or delete every project artifact.

Resources you used:

• The scripts seed_data.py, order_etl.py, and make_bad_encoding.py inside ecommerce-order-etl.
• The data folder containing the CSV, JSON, and JSONL source files.
• The data_bad_encoding folder containing the order feed with the invalid UTF-8 byte.
• The output folder containing accepted orders, rejected orders, and the run summary.

Keep everything running

No action is needed. Choose this if you want to rerun the ETL or continue practising interview scenarios.

• Leave the ecommerce-order-etl workspace in its current location.
• Keep the generated source files so you can compare valid records, record-level failures, and source-level encoding failures.
• Keep the output artifacts as evidence of strict failure, resilient processing, and summary creation.

Pause - I'll come back to this later

There are no background services to stop. Your local files can stay on disk without creating charges.

• Close the terminal panel you used to run the ETL.
• Close the ecommerce-order-etl workspace when you finish reviewing the artifacts.

Your scripts, source data, and generated outputs remain available for your next practice session.

Delete - I don't want to use this again

This option permanently removes the complete local lab. The deletion command targets only the ecommerce-order-etl workspace.

• Copy any artifacts you want to keep to a location outside ecommerce-order-etl.
• Move outside the workspace and delete it by running these commands:

cd ..
rm -rf ecommerce-order-etl

What Do These Commands Remove?

The first line moves the terminal outside ecommerce-order-etl. The second line removes the entire workspace.

This clears the scripts, both source-data folders, and every generated output in one cleanup.

• Check the folder where you saved ecommerce-order-etl using your Linux file manager.

You should no longer see the ecommerce-order-etl workspace.

Still See the Workspace?

• Confirm that your terminal moved to the folder containing ecommerce-order-etl before you retry the deletion.
• Use your Linux file manager to delete the workspace if the terminal reports a permissions problem.
• Help me safely remove my local ecommerce-order-etl workspace.

Nice Work!

Nice Work!

Mission accomplished! You've built a runnable e-commerce ETL that turns file-based orders into traceable outcomes.

What you learned:

• Load a CSV catalog into a product lookup. Turn a JSON customer document into a customer lookup. Apply explicit UTF-8 handling at both file boundaries.
• Stream a JSONL order feed one line at a time. Write accepted orders to CSV. Preserve rejected records in JSONL. Capture every run outcome in a JSON summary.
• Run the pipeline in strict mode. Run it again in resilient mode. Compare fail-fast processing with record quarantine through the output artifacts. Trace exception chaining from low-level errors to domain-specific failures. Use context managers for file cleanup. Use finally for run-level reporting.
• Secret Mission: Completed an optional extension that strengthens source failure handling.

Ready to quiz yourself?