🔵 Intermediate

JSON & CSV Handling

📖 Lesson 25 ⏱ 35 min 🧪 5 questions 💻 3 exercises

🎯 Learning Objectives

Read and write JSON files and strings with the json module
Handle custom types, pretty-printing, and encoding edge cases
Read and write CSV files with csv.reader/writer and DictReader/DictWriter
Handle CSV dialects, quoting, and encoding correctly
Transform and filter data between JSON and CSV formats
Apply best practices for both formats in production code

Why JSON and CSV?

JSON and CSV are the two most common data interchange formats you'll encounter as a Python developer — APIs speak JSON, spreadsheets and databases export CSV. Python's standard library handles both with zero dependencies.

	JSON	CSV
Best for	Nested / hierarchical data, APIs, config	Tabular data, spreadsheet export, bulk records
Human-readable	Yes (with indent)	Yes
Supports types	str, int, float, bool, null, list, dict	Strings only (you parse types yourself)
Nesting	Unlimited depth	Flat (one level)
Python module	`json`	`csv`

JSON Basics

Reading JSON

import json

# ── From a string ──
json_str = '{"name": "Alice", "age": 30, "active": true, "score": null}'
data = json.loads(json_str)

print(data["name"])    # Alice
print(data["active"])  # True   — JSON true → Python True
print(data["score"])   # None   — JSON null → Python None
print(type(data))      # <class 'dict'>

# ── From a file ──
with open("user.json", "r", encoding="utf-8") as f:
    data = json.load(f)

json_read.py

Writing JSON

import json

data = {
    "name":      "Alice",
    "age":       30,
    "languages": ["Python", "Rust"],
    "address":   {"city": "London", "postcode": "EC1A 1BB"},
    "active":    True,
    "balance":   None,
}

# ── To a string ──
compact    = json.dumps(data)                      # single line
pretty     = json.dumps(data, indent=2)            # indented
sorted_str = json.dumps(data, indent=2, sort_keys=True)

print(pretty)

# ── To a file ──
with open("user.json", "w", encoding="utf-8") as f:
    json.dump(data, f, indent=2, ensure_ascii=False)

json_write.py

Always pass ensure_ascii=False when your data contains non-ASCII characters (accented letters, emoji, CJK). Without it, Python escapes them as \uXXXX sequences — valid JSON, but unreadable in a text editor.

Type mapping

Python	JSON	Python ← JSON
`dict`	`{}` object	`dict`
`list`, `tuple`	`[]` array	`list`
`str`	string	`str`
`int`	number (integer)	`int`
`float`	number (decimal)	`float`
`True` / `False`	`true` / `false`	`bool`
`None`	`null`	`None`

tuple serialises as a JSON array, but deserialises back as a list — you lose the tuple type on the round-trip. Similarly, dict keys are always strings in JSON; integer keys are coerced to strings on serialisation.

Custom JSON Encoding & Decoding

Encoding unsupported types

import json
from datetime import datetime, date
from decimal import Decimal
from pathlib import Path

# ── Option 1: custom encoder subclass ──
class AppEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        if isinstance(obj, Path):
            return str(obj)
        if hasattr(obj, "__dict__"):
            return obj.__dict__   # plain object → dict
        return super().default(obj)   # raises TypeError for unknown types

data = {
    "created":  datetime(2024, 3, 15, 10, 30),
    "price":    Decimal("9.99"),
    "log_path": Path("/var/log/app.log"),
}

print(json.dumps(data, cls=AppEncoder, indent=2))

# ── Option 2: default= function (simpler for one-off cases) ──
def encode(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Not serialisable: {type(obj)}")

print(json.dumps({"ts": datetime.now()}, default=encode))

custom_encoder.py

Custom decoding

import json
from datetime import datetime

def decode_dates(dct):
    """Convert ISO date strings back to datetime objects during decode."""
    for key, value in dct.items():
        if isinstance(value, str):
            try:
                dct[key] = datetime.fromisoformat(value)
            except ValueError:
                pass
    return dct

json_str = '{"name": "Alice", "created": "2024-03-15T10:30:00"}'
data = json.loads(json_str, object_hook=decode_dates)

print(data["created"])         # 2024-03-15 10:30:00
print(type(data["created"]))   # <class 'datetime.datetime'>

custom_decoder.py

JSON Best Practices

import json
from pathlib import Path

# ── Atomic JSON write — prevents corrupt files on crash ──
def save_json(path, data, **kwargs):
    tmp = Path(str(path) + ".tmp")
    try:
        tmp.write_text(
            json.dumps(data, ensure_ascii=False, indent=2, **kwargs),
            encoding="utf-8"
        )
        tmp.replace(path)
    except Exception:
        tmp.unlink(missing_ok=True)
        raise

# ── Safe read with default ──
def load_json(path, default=None):
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return default

# ── Validate structure after loading ──
def load_config(path):
    cfg = load_json(path, default={})
    required = {"host", "port", "debug"}
    missing = required - cfg.keys()
    if missing:
        raise ValueError(f"Config missing required keys: {missing}")
    return cfg

json_best_practices.py

CSV Basics

Writing CSV

import csv

rows = [
    ["name",  "department",   "salary"],
    ["Alice", "Engineering",  95000],
    ["Bob",   "Marketing",    78000],
    ["Carol", "Engineering", 102000],
]

# Always open with newline="" — let the csv module handle line endings
with open("employees.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerows(rows)

# ── DictWriter — write dicts with header ──
employees = [
    {"name": "Alice", "dept": "Engineering", "salary": 95000},
    {"name": "Bob",   "dept": "Marketing",   "salary": 78000},
]

with open("employees.csv", "w", newline="", encoding="utf-8") as f:
    fields = ["name", "dept", "salary"]
    writer = csv.DictWriter(f, fieldnames=fields)
    writer.writeheader()
    writer.writerows(employees)

csv_write.py

Reading CSV

import csv

# ── csv.reader — rows as lists ──
with open("employees.csv", "r", newline="", encoding="utf-8") as f:
    reader = csv.reader(f)
    header = next(reader)        # consume header row
    print("Columns:", header)
    for row in reader:
        print(row)               # ['Alice', 'Engineering', '95000']

# ── csv.DictReader — rows as dicts ──
with open("employees.csv", "r", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)   # header row auto-consumed
    for row in reader:
        print(f"{row['name']} in {row['dept']} earns £{int(row['salary']):,}")

csv_read.py

Always pass newline="" when opening CSV files. The csv module does its own universal newline translation. Without newline="", Python's text mode also translates \r\n, causing double blank rows on Windows.

Dialects, Quoting, and Delimiters

import csv

# ── Custom delimiter (TSV — tab-separated) ──
with open("data.tsv", "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f, delimiter="\t")
    writer.writerows([["Alice", 30], ["Bob", 25]])

# ── Quoting options ──
data = [["name", "bio"], ["Alice", 'Says "hello, world"']]

with open("quoted.csv", "w", newline="", encoding="utf-8") as f:
    # QUOTE_ALL — quote every field
    writer = csv.writer(f, quoting=csv.QUOTE_ALL)
    writer.writerows(data)

# ── Registering a custom dialect ──
csv.register_dialect(
    "pipes",
    delimiter="|",
    quotechar='"',
    quoting=csv.QUOTE_MINIMAL,
    lineterminator="\n",
)

with open("pipe_data.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f, dialect="pipes")
    writer.writerows([["Alice", "Engineering"], ["Bob", "Marketing"]])

# ── Sniff dialect from an existing file ──
with open("unknown.csv", "r", newline="", encoding="utf-8") as f:
    sample  = f.read(1024)
    dialect = csv.Sniffer().sniff(sample)
    f.seek(0)
    reader = csv.reader(f, dialect)
    for row in reader:
        print(row)

dialects.py

Type Conversion in CSV

CSV is always text — every field comes back as a string. You must convert types yourself:

import csv
from datetime import date

def parse_employee(row):
    """Convert a CSV row dict to typed Python values."""
    return {
        "name":       row["name"].strip().title(),
        "salary":     int(row["salary"]),
        "start_date": date.fromisoformat(row["start_date"]),
        "active":     row["active"].strip().lower() == "true",
        "score":      float(row["score"]) if row["score"] else None,
    }

csv_content = """name,salary,start_date,active,score
alice,95000,2020-03-01,True,9.5
bob,78000,2019-11-15,False,
carol,102000,2021-07-20,True,8.8
"""

import io
reader = csv.DictReader(io.StringIO(csv_content))
employees = [parse_employee(row) for row in reader]
for e in employees:
    print(e)

csv_types.py

JSON ↔ CSV Conversion

import json, csv, io

# ── JSON array of objects → CSV ──
json_data = [
    {"name": "Alice", "dept": "Engineering", "salary": 95000},
    {"name": "Bob",   "dept": "Marketing",   "salary": 78000},
    {"name": "Carol", "dept": "Engineering", "salary": 102000},
]

def json_to_csv(records, path):
    if not records:
        return
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=records[0].keys())
        writer.writeheader()
        writer.writerows(records)

json_to_csv(json_data, "output.csv")

# ── CSV → JSON array of objects ──
def csv_to_json(csv_path, json_path, type_map=None):
    """type_map: dict of {column: callable} for type conversions."""
    type_map = type_map or {}
    with open(csv_path, newline="", encoding="utf-8") as f:
        records = list(csv.DictReader(f))
    for row in records:
        for col, convert in type_map.items():
            if col in row and row[col]:
                row[col] = convert(row[col])
    with open(json_path, "w", encoding="utf-8") as f:
        json.dump(records, f, indent=2, ensure_ascii=False)

csv_to_json("output.csv", "output.json", type_map={"salary": int})

conversion.py

Streaming Large Files

For multi-GB CSV files, never load everything into memory — iterate row by row:

import csv
from collections import defaultdict

def summarise_csv(path, group_col, value_col):
    """
    Group by group_col and sum value_col.
    Processes one row at a time — works for any file size.
    """
    totals = defaultdict(float)
    count  = 0

    with open(path, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            try:
                totals[row[group_col]] += float(row[value_col])
                count += 1
            except (ValueError, KeyError):
                pass   # skip malformed rows

    print(f"Processed {count:,} rows")
    return dict(totals)

# result = summarise_csv("transactions.csv", "department", "amount")

streaming.py

Error Handling

import json, csv

# ── JSON errors ──
def safe_json_loads(text, default=None):
    try:
        return json.loads(text)
    except json.JSONDecodeError as e:
        print(f"JSON parse error at line {e.lineno}, col {e.colno}: {e.msg}")
        return default

print(safe_json_loads('{"valid": true}'))   # {'valid': True}
print(safe_json_loads('{broken json}'))     # None + error message

# ── CSV errors — handle malformed rows gracefully ──
def read_csv_safe(path):
    records = []
    errors  = 0
    with open(path, newline="", encoding="utf-8", errors="replace") as f:
        reader = csv.DictReader(f)
        for line_num, row in enumerate(reader, start=2):
            try:
                records.append({
                    "name":   row["name"].strip(),
                    "salary": int(row["salary"]),
                })
            except (ValueError, KeyError) as e:
                print(f"Skipping line {line_num}: {e}")
                errors += 1
    print(f"Loaded {len(records)} records, skipped {errors} errors")
    return records

error_handling.py

Best Practices

Rule	JSON	CSV
Always specify encoding	`ensure_ascii=False` + `encoding="utf-8"`	`encoding="utf-8"`
File opening	Normal text mode	Always `newline=""`
Pretty print	`indent=2`	N/A
Large files	Use `ijson` for streaming	Iterate rows — never `readlines()`
Type safety	Types preserved (mostly)	Always convert manually after reading
Atomicity	Write to `.tmp`, then rename	Write to `.tmp`, then rename
Validation	Check required keys after loading	Validate and skip malformed rows

Primary sources: Python Docs — json · Python Docs — csv · Python Docs — csv.Sniffer

🤖

Ask your AI tutor! Getting a JSONDecodeError from an API response? CSV rows coming back with wrong types? Need to handle a non-standard delimiter or quoting style? Great problems to debug together.

💻 Exercises

01 JSON Config Manager

Build a JsonConfig class that:

Loads a JSON file on init (creates it with defaults if absent)
get(key, default=None) — retrieve a value
set(key, value) — update a value in memory
save() — write atomically to the file
Handles datetime values by encoding as ISO strings and decoding back automatically

Show solution

import json
from pathlib import Path
from datetime import datetime

class DateTimeEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return {"__datetime__": obj.isoformat()}
        return super().default(obj)

def datetime_decoder(dct):
    for k, v in dct.items():
        if isinstance(v, dict) and "__datetime__" in v:
            dct[k] = datetime.fromisoformat(v["__datetime__"])
    return dct

class JsonConfig:
    def __init__(self, path, defaults=None):
        self.path = Path(path)
        self._data = dict(defaults or {})
        if self.path.exists():
            with self.path.open(encoding="utf-8") as f:
                loaded = json.load(f, object_hook=datetime_decoder)
            self._data.update(loaded)
        else:
            self.save()

    def get(self, key, default=None):
        return self._data.get(key, default)

    def set(self, key, value):
        self._data[key] = value

    def save(self):
        tmp = Path(str(self.path) + ".tmp")
        try:
            tmp.write_text(
                json.dumps(self._data, cls=DateTimeEncoder, indent=2),
                encoding="utf-8"
            )
            tmp.replace(self.path)
        except Exception:
            tmp.unlink(missing_ok=True)
            raise

    def __repr__(self):
        return f"JsonConfig({self.path}, {self._data})"

# Test
cfg = JsonConfig("app.json", defaults={"debug": False, "port": 8080})
cfg.set("last_run", datetime.now())
cfg.set("port", 9000)
cfg.save()

cfg2 = JsonConfig("app.json")
print(cfg2.get("port"))      # 9000
print(type(cfg2.get("last_run")))  # datetime

02 CSV Report Generator

Given this CSV data (create it programmatically):

date,product,region,units,unit_price
2024-01-15,Widget,North,150,9.99
2024-01-15,Gadget,South,80,24.99
2024-01-16,Widget,South,200,9.99
2024-01-16,Gadget,North,45,24.99
2024-01-17,Widget,North,175,9.99
2024-01-17,Gadget,South,90,24.99

Write a function sales_report(csv_path) that reads the CSV and prints:

Total revenue per product (units × unit_price)
Total revenue per region
The single best-selling day (highest total revenue)

Show solution

import csv
from collections import defaultdict
from pathlib import Path

# Create test data
Path("sales.csv").write_text(
    "date,product,region,units,unit_price\n"
    "2024-01-15,Widget,North,150,9.99\n"
    "2024-01-15,Gadget,South,80,24.99\n"
    "2024-01-16,Widget,South,200,9.99\n"
    "2024-01-16,Gadget,North,45,24.99\n"
    "2024-01-17,Widget,North,175,9.99\n"
    "2024-01-17,Gadget,South,90,24.99\n",
    encoding="utf-8"
)

def sales_report(csv_path):
    by_product = defaultdict(float)
    by_region  = defaultdict(float)
    by_date    = defaultdict(float)

    with open(csv_path, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            revenue = int(row["units"]) * float(row["unit_price"])
            by_product[row["product"]] += revenue
            by_region[row["region"]]   += revenue
            by_date[row["date"]]       += revenue

    print("── Revenue by product ──")
    for product, total in sorted(by_product.items()):
        print(f"  {product:<10} £{total:>10,.2f}")

    print("\n── Revenue by region ──")
    for region, total in sorted(by_region.items()):
        print(f"  {region:<10} £{total:>10,.2f}")

    best_day, best_rev = max(by_date.items(), key=lambda x: x[1])
    print(f"\n── Best day: {best_day} with £{best_rev:,.2f} ──")

sales_report("sales.csv")

03 JSON ↔ CSV Round-Trip

Write two functions:

flatten_record(record) — takes a nested dict (one level of nesting) and returns a flat dict suitable for CSV (e.g. {"address": {"city": "London"}} → {"address.city": "London"})
unflatten_record(flat) — reverses the process

Use these to convert a list of nested JSON records to CSV and back, verifying the round-trip produces identical data.

Show solution

import csv, json, io

def flatten_record(record, sep="."):
    """Flatten one level of nesting: {a: {b: 1}} → {a.b: 1}"""
    flat = {}
    for key, value in record.items():
        if isinstance(value, dict):
            for sub_key, sub_val in value.items():
                flat[f"{key}{sep}{sub_key}"] = sub_val
        else:
            flat[key] = value
    return flat

def unflatten_record(flat, sep="."):
    """Restore one level of nesting: {a.b: 1} → {a: {b: 1}}"""
    result = {}
    for key, value in flat.items():
        if sep in key:
            parent, child = key.split(sep, 1)
            result.setdefault(parent, {})[child] = value
        else:
            result[key] = value
    return result

# Test data
records = [
    {"name": "Alice", "age": 30, "address": {"city": "London",     "postcode": "EC1A 1BB"}},
    {"name": "Bob",   "age": 25, "address": {"city": "Manchester", "postcode": "M1 1AE"}},
]

# Flatten → CSV → unflatten
flat_records = [flatten_record(r) for r in records]
print("Flat:", flat_records[0])

buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=flat_records[0].keys())
writer.writeheader()
writer.writerows(flat_records)

buf.seek(0)
restored = [unflatten_record(row) for row in csv.DictReader(buf)]

# Verify round-trip (note: CSV turns int age into str)
for orig, rest in zip(records, restored):
    rest["age"] = int(rest["age"])   # restore type
    assert orig == rest, f"Mismatch: {orig} != {rest}"
print("Round-trip verified ✓")