JSON Automation: Validate & Transform

By the end of this lesson you can:

• Load and dump JSON files with json.load/json.dump (and the string forms).
• Reach into nested structures safely with .get and defaults.
• Validate that incoming JSON has the keys and types you expect.
• Transform a nested document into a flat shape (e.g. ready for CSV).

JSON maps almost perfectly onto Python types you already know:

JSON          Python
object {}  →  dict
array []   →  list
string     →  str
number     →  int / float
true/false →  True / False
null       →  None

Loading and dumping

import json
from pathlib import Path

# from a file
data = json.loads(Path("config.json").read_text(encoding="utf-8"))

# back to a file, pretty-printed
Path("out.json").write_text(
    json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")

• json.loads(s) parses a string; json.load(file) reads from a file object. (The s = "string".)
• json.dumps(obj, indent=2) makes readable output; ensure_ascii=False keeps emoji and accents as themselves.

Reaching in safely

user = {
    "name": "Aisha",
    "address": {"city": "Kuala Lumpur"},
    "roles": ["admin", "editor"],
}

# risky: KeyError if 'address' or 'zip' is missing
# zip_code = user["address"]["zip"]

# safe: .get with defaults, never crashes
city = user.get("address", {}).get("city", "unknown")
zip_code = user.get("address", {}).get("zip", "—")
first_role = (user.get("roles") or ["none"])[0]
print(city, zip_code, first_role)

External JSON is unreliable — keys vanish, values are null. Chaining .get(key, default) walks the tree without ever raising KeyError. The .get("address", ) trick gives you an empty dict to keep chaining on.

Validating the shape

def validate(record: dict) -> list[str]:
    errors = []
    if not isinstance(record.get("name"), str) or not record["name"]:
        errors.append("name must be a non-empty string")
    if not isinstance(record.get("age"), int):
        errors.append("age must be an integer")
    if not isinstance(record.get("emails"), list):
        errors.append("emails must be a list")
    return errors

bad = {"name": "", "age": "old", "emails": "a@b.com"}
print(validate(bad))
# ['name must be a non-empty string', 'age must be an integer', 'emails must be a list']

A validator that returns a list of problems (rather than raising at the first) lets you report everything wrong at once — far friendlier for whoever sends you the data. For big or shared schemas, libraries like pydantic (Level 8) or jsonschema automate this; the hand-rolled version is plenty for small jobs.

Transforming / reshaping

# nested API response → flat rows for a CSV
api = {
    "users": [
        {"id": 1, "name": "Aisha", "address": {"city": "KL"}},
        {"id": 2, "name": "Ben",   "address": {"city": "London"}},
    ]
}

flat = [
    {"id": u["id"], "name": u["name"],
     "city": u.get("address", {}).get("city", "")}
    for u in api["users"]
]
print(flat)
# [{'id': 1, 'name': 'Aisha', 'city': 'KL'}, {'id': 2, ...}]

Flattening nested JSON into a list of flat dicts is the bridge to Lesson 15's DictWriter — JSON in, CSV out. The comprehension reads as "for each user, build the flat record I want."

Goal: take a messy nested JSON export, validate each record, flatten the good ones to CSV, and log the bad ones — a realistic "clean the export" job.

import json, csv, logging
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("json2csv")

def flatten(user: dict) -> dict:
    return {
        "id":    user.get("id", ""),
        "name":  user.get("name", "").strip(),
        "email": (user.get("contact", {}).get("email") or "").lower(),
        "city":  user.get("address", {}).get("city", ""),
        "roles": "|".join(user.get("roles", [])),   # list → pipe-joined string
    }

def is_valid(user: dict) -> bool:
    return bool(user.get("id")) and "@" in (
        user.get("contact", {}).get("email") or "")

raw = json.loads(Path("export.json").read_text(encoding="utf-8"))
users = raw.get("users", [])

good, bad = [], 0
for u in users:
    if is_valid(u):
        good.append(flatten(u))
    else:
        bad += 1
        log.warning("skipping invalid record id=%s", u.get("id", "?"))

with open("users.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.DictWriter(f, fieldnames=["id", "name", "email", "city", "roles"])
    w.writeheader(); w.writerows(good)

log.info("wrote %d users, skipped %d invalid", len(good), bad)

WARNING skipping invalid record id=7
WARNING skipping invalid record id=?
INFO wrote 48 users, skipped 2 invalid

Read the code

Two small functions carry the logic: flatten reshapes one nested user into a flat row (note roles, a list, becomes a pipe-joined string for CSV), and is_valid guards the boundary. Every reach uses .get(..., default) so a missing key never crashes the batch — one bad record doesn't sink the other 48. This JSON→validate→flatten→CSV pipeline is one of the most common automations you'll ever write.

Parsed JSON is just nested dicts and lists. Load with json.loads/load, dump with json.dumps(obj, indent=2, ensure_ascii=False). Reach into external data with chained .get(key, default) so missing keys never crash you. Validate at the boundary — a function returning a list of problems reports everything at once — and reshape nested documents into the flat structure your code (or a CSV) needs. JSON→validate→flatten→CSV is a workhorse pipeline; for big schemas, lean on pydantic/jsonschema later.