Files, Paths & Payload Ingestion

Ingestion means reading data from outside into your program. A payload is the data a file or another system sends you.

You do not control that data. It can have odd lines, strange letters or a broken ending. Your job is to read it and lose nothing.

You know with open(...) and try/except from earlier chapters. Here is the shop file again. One line is one order:

order_id,customer_id,status,amount
101,1,paid,40.0
102,1,paid,25.0
103,2,refunded,60.0
104,3,paid,15.5
105,3,pending,
106,4,paid,80.0

This file is clean. Real files are not. Each section shows one problem, in a small file, and its fix.

What you will learn

1. How to name a file with pathlib, and how folders show the state of a file.
2. How a file turns from bytes into text, and what an encoding is.
3. Why split(",") breaks, and what the csv module does instead.
4. How to handle rows with too few or too many values.
5. How to read JSON and NDJSON, and report every bad line with its line number.
6. How to decide if a problem is one row, one file, or the whole run.

A path says where a file is. The pathlib module gives you a Path object for it. A Path knows its own parts.

Here the / sign does not divide. A Path uses it to join parts. Then you ask for the name, stem, suffix and parent:

from pathlib import Path

path = Path("data") / "incoming" / "orders.csv"

print(path)
print(path.name)
print(path.stem)
print(path.suffix)
print(path.parent)

A Path can also read the file. exists() checks that the file is there. read_text() returns the whole file as one text:

from pathlib import Path

path = Path("orders.csv")

print(path.exists())
text = path.read_text(encoding="utf-8")
print(text.splitlines()[1])

read_text() loads the whole file into memory. For a large file, use with open(...) and loop over the lines, as in the streaming chapter.

A pipeline needs a promise about its folders. The folder says what state a file is in. A common layout has three folders:

incoming/    files exactly as they arrived
rejected/    bad rows and quarantined files
archive/     files that are finished

Big jobs often add one folder per day, such as incoming/2026-06-19/. This is a partition: one slice of the data, kept in its own folder.

Before you read a file, decide what to do with it. Look at this folder. The next code gives each file one of three answers:

from pathlib import Path

def classify(path):
    if path.suffix != ".csv":
        return "ignore"
    if path.stat().st_size == 0:
        return "quarantine"
    return "read"

for path in sorted(Path("incoming").iterdir()):
    print(path.name, classify(path))

A file stores bytes, which are numbers. An encoding is the rule that turns bytes into letters. The most common rule is utf-8.

The same letter can be different bytes in different encodings. Look at the letter é:

print("café".encode("utf-8"))
print("café".encode("latin-1"))

Here is orders_latin.csv. Its second line has a note, saved in latin-1.

Read it as utf-8. Python stops and shows this error:

with open("orders_latin.csv", encoding="utf-8") as file:
    print(file.read())

You fix it by giving the right encoding. Ask the sender which one the file uses:

with open("orders_latin.csv", encoding="latin-1") as file:
    print(file.read().splitlines()[1])

Always write encoding= yourself. Without it, Python uses the default of the computer. The same code then works on your laptop and fails on the server.

You can cut a line at every comma with split(","). This works on orders.csv. It fails when a value contains a comma.

CSV handles this with quotes. A value in double quotes may hold commas. Here is orders_notes.csv, with a note column:

order_id,customer_id,status,amount,note
101,1,paid,40.0,"gift, wrap it"
102,1,paid,25.0,

next(file) gives the next line of the file. The first call skips the header, and the second gives line 2.

Now split that line at every comma:

with open("orders_notes.csv", encoding="utf-8") as file:
    next(file)
    line = next(file).strip()

parts = line.split(",")
print(parts)
print(len(parts))

The values after the comma now sit in the wrong columns. No error appears. You only see the damage later, in the data.

The csv module knows the quote rules. Open the file with newline="". Then csv can handle line breaks inside quoted values by itself.

csv.reader gives you each line as a list of text values. Try it on the file that broke split:

import csv

with open("orders_notes.csv", encoding="utf-8", newline="") as file:
    reader = csv.reader(file)
    for row in reader:
        print(row)

csv.DictReader reads the header for you. It gives each row as a dict, with the header names as keys. This is the form you know from the records chapter:

import csv

with open("orders.csv", encoding="utf-8", newline="") as file:
    for row in csv.DictReader(file):
        print(row)

DictReader uses names, not positions. If the sender adds a column in the middle, row["amount"] still works. With csv.reader, row[3] would now be the wrong value.

Some tools, such as Excel, start a utf-8 file with three hidden bytes. These bytes are a BOM (byte order mark).

The next file, orders_bom.csv, has a BOM. Read it with DictReader and plain utf-8:

import csv

with open("orders_bom.csv", encoding="utf-8", newline="") as file:
    row = next(csv.DictReader(file))

print(row)
print(row.get("order_id"))

The fix is the encoding utf-8-sig. It drops the BOM, and it also works on files that have none:

import csv

with open("orders_bom.csv", encoding="utf-8-sig", newline="") as file:
    row = next(csv.DictReader(file))

print(row)

A ragged row has a different number of values than the header. A tool dropped a value, or added one.

Here is orders_messy.csv. Its header has 4 names. The table counts the values on each line:

order_id,customer_id,status,amount
101,1,paid,40.0
102,1,paid

103,2,refunded,60.0
104,3,paid,abc
105,3,pending,
106,4,paid,80.0,extra

See what DictReader does with these rows:

import csv

with open("orders_messy.csv", encoding="utf-8", newline="") as file:
    for row in csv.DictReader(file):
        print(row)

The None of order 102 differs from the empty text of order 105. Both mean "no amount". But only 102 is a broken line, because it has 3 values, not 4.

To catch this, compare the length of each row with the header. csv.reader has a line_num that tells you the current line:

import csv

with open("orders_messy.csv", encoding="utf-8", newline="") as file:
    reader = csv.reader(file)
    header = next(reader)
    for row in reader:
        if not row:
            continue
        if len(row) != len(header):
            print(
                "line", reader.line_num,
                "has", len(row), "values, header has", len(header),
            )

Next you want a dict for each good row. zip(header, row) pairs each name with its value. Then dict() makes a dict:

header = ["order_id", "customer_id", "status", "amount"]

print(dict(zip(header, ["101", "1", "paid", "40.0"])))
print(dict(zip(header, ["102", "1", "paid"])))
print(dict(zip(header, ["106", "4", "paid", "80.0", "extra"])))

A row that fails a check is a bad row. You must not drop it in silence. Keep two lists: good rows, and bad rows with a line number and a reason.

This function reads orders_messy.csv. It rejects a row with the wrong length, and a row whose amount is not a number. An empty amount is allowed, as in order 105:

import csv

def read_orders(path):
    good = []
    bad = []
    seen = 0
    # utf-8-sig also works on files without a BOM
    with open(path, encoding="utf-8-sig", newline="") as file:
        reader = csv.reader(file)
        header = next(reader)
        for row in reader:
            if not row:
                continue  # a blank line holds no data
            seen += 1
            line = reader.line_num
            if len(row) != len(header):
                reason = f"{len(row)} values, header has {len(header)}"
                bad.append({"line": line, "reason": reason})
                continue
            order = dict(zip(header, row))
            amount = order["amount"]
            if amount == "":
                order["amount"] = None
            else:
                try:
                    order["amount"] = float(amount)
                except ValueError:
                    reason = f"amount is not a number: {amount!r}"
                    bad.append({"line": line, "reason": reason})
                    continue
            good.append(order)
    return good, bad, seen

good, bad, seen = read_orders("orders_messy.csv")
print("good:", [order["order_id"] for order in good])
for item in bad:
    print(item)
print("rows seen:", seen)
print("good + bad:", len(good) + len(bad))

Here is what the function did:

• A row with the wrong length goes to the bad list. The code never calls zip on it.
• float(amount) turns text into a number. For text such as abc it raises ValueError, and the except part catches it.
• {amount!r} shows the value with quotes, so you see exactly what the file held.
• It returns three things: the good list, the bad list, and the count of rows seen.

seen counts every row a second way, in the loop. Good plus bad must equal seen. If they differ, your code lost a row.

JSON is a text format for nested data. Nested means a value holds other values, such as a list inside an object. A JSON object looks like a Python dict.

json.loads turns JSON text into Python objects. Look at the types. A JSON number stays a number, and JSON null becomes None:

import json

text = '{"order_id": 101, "customer_id": 1, "status": "paid", "amount": 40.0}'
order = json.loads(text)
print(order)
print(type(order).__name__, order["amount"])

pending = json.loads('{"order_id": 105, "amount": null}')
print(pending)

Here is orders.json. It holds a list of orders under the key "orders":

{"orders": [
  {"order_id": 101, "customer_id": 1, "status": "paid", "amount": 40.0},
  {"order_id": 105, "customer_id": 3, "status": "pending", "amount": null}
]}

Read the file with read_text(), then json.loads. (json.load reads from an open file. json.loads reads from a text.) Then use the key "orders":

import json
from pathlib import Path

text = Path("orders.json").read_text(encoding="utf-8")
data = json.loads(text)

print(list(data))
for order in data["orders"]:
    print(order["order_id"], order["amount"])

If the sender changes the format, json.loads may return a list, not a dict. Check before you use it. isinstance(payload, dict) is True only for a dict:

import json

payload = json.loads('[{"order_id": 101}]')

if not isinstance(payload, dict):
    print("rejected: expected an object, got", type(payload).__name__)

A JSON document is read as a whole. One wrong character, and nothing is read. Here the closing brace is missing:

import json

try:
    json.loads('{"order_id": 101')
except json.JSONDecodeError as error:
    print(error)

NDJSON means newline-delimited JSON: JSON objects separated by line breaks. Each line is one object. You can read one line at a time, like a CSV.

Here is orders.ndjson. Line 4 is broken. It has no closing brace:

{"order_id": 101, "customer_id": 1, "status": "paid", "amount": 40.0}
{"order_id": 102, "customer_id": 1, "status": "paid", "amount": 25.0}
{"order_id": 103, "customer_id": 2, "status": "refunded", "amount": 60.0}
{"order_id": 104, "customer_id": 3, "status": "paid", "amount": 15.5
{"order_id": 105, "customer_id": 3, "status": "pending", "amount": null}
{"order_id": 106, "customer_id": 4, "status": "paid", "amount": 80.0}

First try to read the whole file as one JSON document:

import json

try:
    json.load(open("orders.ndjson", encoding="utf-8"))
except json.JSONDecodeError as error:
    print(error)

Now read it line by line. enumerate(file, start=1) gives a count that starts at 1, with each line. Each line gets its own try/except, so a bad line is skipped:

import json

good = []
bad = []
with open("orders.ndjson", encoding="utf-8") as file:
    for line_number, line in enumerate(file, start=1):
        if not line.strip():
            continue  # skip a blank line
        try:
            order = json.loads(line)
        except json.JSONDecodeError as error:
            bad.append({"line": line_number, "reason": error.msg})
            continue
        good.append(order)

print([order["order_id"] for order in good])
print(bad)

A line can be valid JSON but not an object, such as [1, 2]. Reject it with a reason, as in the shape check.

Not every problem has the same size. A run is one start of your job, over all its files. Choose the smallest fix that keeps the output safe.

A missing column in one row is a row problem. A missing column in the header is a file problem, because every row is affected. Check the header first:

header = ["order_id", "status", "amount"]
required = {"order_id", "customer_id", "status", "amount"}

missing = required - set(header)
print(missing)

A try/except around the whole loop is too big. One bad line would then stop thousands of good rows. Put the try inside the loop, as in the NDJSON code.

You learned to name files, read CSV and JSON safely, and report every bad line. Each exercise opens in the Python workspace with its own data.

Before you run an exercise, guess which line of the input will cause trouble. Then check your guess.