CSV, JSON, and JSONL: Structured Text Done Right
1 · The lesson
readThree formats cover most of the structured text you'll work with in Python: CSV for tabular data, JSON for nested objects, JSONL for streaming. The standard library handles all three well. It also has a few traps, and almost everyone gets caught by newline="" or json.dumps(datetime.now()) at some point.
This lesson is a practical reference for those three formats, plus a short section on TOML and YAML for when you need to store config, not data.
1. CSV: Reading, and the One Argument Everyone Forgets
The csv module reads and writes comma-separated (and tab-separated, and pipe-separated) text files. The headline API:
import csv # Read as lists with open("people.csv", encoding="utf-8", newline="") as f: reader = csv.reader(f) header = next(reader) # skip the header row for row in reader: print(row) # ['Alice', '30', 'alice@example.com'] # Read as dicts, which is almost always what you want with open("people.csv", encoding="utf-8", newline="") as f: for row in csv.DictReader(f): print(row["name"], row["email"])
That newline="" is the argument people forget most often. The csv module does its own newline handling (necessary because fields can contain \n inside quotes), and Python's default text-mode newline translation interferes with it. On Windows especially, if you leave out newline="" when writing, you get a blank row after every real row. On Linux and Mac it usually works by luck, and then your script breaks when a Windows user runs it.
Rule: every open() of a CSV file gets newline="", for reading and for writing.
2. CSV: Writing
The write side mirrors the read side. csv.writer for lists, csv.DictWriter for dicts.
import csv from pathlib import Path rows = [ {"name": "Alice", "age": 30, "email": "alice@example.com"}, {"name": "Bob", "age": 25, "email": "bob@example.com"}, ] with open("out.csv", "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=["name", "age", "email"]) writer.writeheader() writer.writerows(rows) # writes all rows at once (.writerow writes one)
fieldnames= is required on purpose: you have to declare your columns. That makes the output stable: same columns, same order, every run. writer.writerows(iterable) is faster than calling writerow in a loop, and shorter to read.
For data that may contain quotes or commas inside fields, the module handles escaping automatically. Don't write CSVs by hand with f.write(",".join(row)). It works for simple data, but as soon as a name contains "O'Brien, Jr.", the file can no longer be parsed correctly.
3. CSV: Quoting, Dialects, and the BOM Problem
The csv module's default is the "excel" dialect: comma-separated, double quotes for quoting, and quotes only when needed. You can change this for each writer:
writer = csv.writer(f, quoting=csv.QUOTE_ALL) # quote every field writer = csv.writer(f, quoting=csv.QUOTE_NONNUMERIC) # quote everything except numbers writer = csv.writer(f, delimiter="\t") # TSV writer = csv.writer(f, delimiter="|") # pipe-separated
setup added so this can run · re-runs 2 earlier blocks on this page first
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) f = _AutoMock('f') csv = _AutoMock('csv') # Earlier blocks on this page that this one builds on, re-run quietly first. try: import csv from pathlib import Path except Exception: pass _pm_earlier = [ 'import csv\n\n# Read as lists\nwith open("people.csv", encoding="utf-8", newline="") as f:\n reader = csv.reader(f)\n header = next(reader) # skip the header row\n for row in reader:\n print(row) # [\'Alice\', \'30\', \'alice@example.com\']\n\n# Read as dicts, which is almost always what you want\nwith open("people.csv", encoding="utf-8", newline="") as f:\n for row in csv.DictReader(f):\n print(row["name"], row["email"])', 'import csv\nfrom pathlib import Path\n\nrows = [\n {"name": "Alice", "age": 30, "email": "alice@example.com"},\n {"name": "Bob", "age": 25, "email": "bob@example.com"},\n]\n\nwith open("out.csv", "w", encoding="utf-8", newline="") as f:\n writer = csv.DictWriter(f, fieldnames=["name", "age", "email"])\n writer.writeheader()\n writer.writerows(rows) # writes all rows at once (.writerow writes one)', ] def _pm_run_earlier(_srcs, _g): import sys as _s, io as _io, warnings as _w for _src in _srcs: _o, _e = _s.stdout, _s.stderr _s.stdout = _s.stderr = _io.StringIO() try: with _w.catch_warnings(): _w.simplefilter('ignore') exec(compile(_src, '<earlier block>', 'exec'), _g) except (Exception, SystemExit): pass finally: _s.stdout, _s.stderr = _o, _e if 'matplotlib.pyplot' in _s.modules: _s.modules['matplotlib.pyplot'].close('all') # their plots belong to their own blocks _pm_run_earlier(_pm_earlier, globals()) del _pm_earlier, _pm_run_earlier
For reading files of unknown dialect, the sniffer guesses:
with open("mystery.csv", encoding="utf-8", newline="") as f: sample = f.read(2048) f.seek(0) dialect = csv.Sniffer().sniff(sample) reader = csv.reader(f, dialect)
setup added so this can run · re-runs 3 earlier blocks on this page first
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) csv = _AutoMock('csv') f = _AutoMock('f') # Earlier blocks on this page that this one builds on, re-run quietly first. try: import csv from pathlib import Path except Exception: pass _pm_earlier = [ 'import csv\n\n# Read as lists\nwith open("people.csv", encoding="utf-8", newline="") as f:\n reader = csv.reader(f)\n header = next(reader) # skip the header row\n for row in reader:\n print(row) # [\'Alice\', \'30\', \'alice@example.com\']\n\n# Read as dicts, which is almost always what you want\nwith open("people.csv", encoding="utf-8", newline="") as f:\n for row in csv.DictReader(f):\n print(row["name"], row["email"])', 'import csv\nfrom pathlib import Path\n\nrows = [\n {"name": "Alice", "age": 30, "email": "alice@example.com"},\n {"name": "Bob", "age": 25, "email": "bob@example.com"},\n]\n\nwith open("out.csv", "w", encoding="utf-8", newline="") as f:\n writer = csv.DictWriter(f, fieldnames=["name", "age", "email"])\n writer.writeheader()\n writer.writerows(rows) # writes all rows at once (.writerow writes one)', 'writer = csv.writer(f, quoting=csv.QUOTE_ALL) # quote every field\nwriter = csv.writer(f, quoting=csv.QUOTE_NONNUMERIC) # quote everything except numbers\nwriter = csv.writer(f, delimiter="\\t") # TSV\nwriter = csv.writer(f, delimiter="|") # pipe-separated', ] def _pm_run_earlier(_srcs, _g): import sys as _s, io as _io, warnings as _w for _src in _srcs: _o, _e = _s.stdout, _s.stderr _s.stdout = _s.stderr = _io.StringIO() try: with _w.catch_warnings(): _w.simplefilter('ignore') exec(compile(_src, '<earlier block>', 'exec'), _g) except (Exception, SystemExit): pass finally: _s.stdout, _s.stderr = _o, _e if 'matplotlib.pyplot' in _s.modules: _s.modules['matplotlib.pyplot'].close('all') # their plots belong to their own blocks _pm_run_earlier(_pm_earlier, globals()) del _pm_earlier, _pm_run_earlier
The BOM problem. CSVs exported from Excel often start with a Unicode Byte-Order Mark (\ufeff). With encoding="utf-8" that BOM ends up as the first character of your first field name, and row["Name"] raises KeyError because the key is really "\ufeffName". Fix it with encoding="utf-8-sig":
# utf-8-sig removes the BOM (byte order mark) that Excel adds with open("excel_export.csv", encoding="utf-8-sig", newline="") as f: for row in csv.DictReader(f): print(row["Name"]) # works now, because the BOM is gone
setup added so this can run · re-runs 4 earlier blocks on this page first
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) csv = _AutoMock('csv') f = _AutoMock('f') # Earlier blocks on this page that this one builds on, re-run quietly first. try: import csv from pathlib import Path except Exception: pass _pm_earlier = [ 'import csv\n\n# Read as lists\nwith open("people.csv", encoding="utf-8", newline="") as f:\n reader = csv.reader(f)\n header = next(reader) # skip the header row\n for row in reader:\n print(row) # [\'Alice\', \'30\', \'alice@example.com\']\n\n# Read as dicts, which is almost always what you want\nwith open("people.csv", encoding="utf-8", newline="") as f:\n for row in csv.DictReader(f):\n print(row["name"], row["email"])', 'import csv\nfrom pathlib import Path\n\nrows = [\n {"name": "Alice", "age": 30, "email": "alice@example.com"},\n {"name": "Bob", "age": 25, "email": "bob@example.com"},\n]\n\nwith open("out.csv", "w", encoding="utf-8", newline="") as f:\n writer = csv.DictWriter(f, fieldnames=["name", "age", "email"])\n writer.writeheader()\n writer.writerows(rows) # writes all rows at once (.writerow writes one)', 'writer = csv.writer(f, quoting=csv.QUOTE_ALL) # quote every field\nwriter = csv.writer(f, quoting=csv.QUOTE_NONNUMERIC) # quote everything except numbers\nwriter = csv.writer(f, delimiter="\\t") # TSV\nwriter = csv.writer(f, delimiter="|") # pipe-separated', 'with open("mystery.csv", encoding="utf-8", newline="") as f:\n sample = f.read(2048)\n f.seek(0)\n dialect = csv.Sniffer().sniff(sample)\n reader = csv.reader(f, dialect)', ] def _pm_run_earlier(_srcs, _g): import sys as _s, io as _io, warnings as _w for _src in _srcs: _o, _e = _s.stdout, _s.stderr _s.stdout = _s.stderr = _io.StringIO() try: with _w.catch_warnings(): _w.simplefilter('ignore') exec(compile(_src, '<earlier block>', 'exec'), _g) except (Exception, SystemExit): pass finally: _s.stdout, _s.stderr = _o, _e if 'matplotlib.pyplot' in _s.modules: _s.modules['matplotlib.pyplot'].close('all') # their plots belong to their own blocks _pm_run_earlier(_pm_earlier, globals()) del _pm_earlier, _pm_run_earlier
utf-8-sig strips the BOM on read. For write, prefer plain utf-8 (most tooling handles BOM-less UTF-8 fine, and writing a BOM annoys downstream tools).
4. JSON: Four Functions
The json module has four functions you need to remember:
| Function | Direction | Operates on |
|---|---|---|
json.dumps(obj) | Python → JSON | string |
json.loads(s) | JSON → Python | string |
json.dump(obj, f) | Python → JSON | file |
json.load(f) | JSON → Python | file |
The s suffix means "string". dump/load work on file objects directly, so there is no need to read the file into a string first.
import json from pathlib import Path config = {"version": 3, "features": ["dark_mode", "autosave"], "limits": {"max": 100}} # String round-trip text = json.dumps(config) # '{"version": 3, ...}' back = json.loads(text) # {'version': 3, ...} # File round-trip with open("config.json", "w", encoding="utf-8") as f: json.dump(config, f, indent=2) with open("config.json", encoding="utf-8") as f: loaded = json.load(f)
For human-readable output, pass indent=2 (or 4). Add sort_keys=True so the key order stays the same and diffs in version control stay small:
json.dumps(config, indent=2, sort_keys=True)
setup added so this can run · re-runs 1 earlier block on this page first
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) config = _AutoMock('config') json = _AutoMock('json') # Earlier blocks on this page that this one builds on, re-run quietly first. try: import json from pathlib import Path except Exception: pass _pm_earlier = [ 'import json\nfrom pathlib import Path\n\nconfig = {"version": 3, "features": ["dark_mode", "autosave"], "limits": {"max": 100}}\n\n# String round-trip\ntext = json.dumps(config) # \'{"version": 3, ...}\'\nback = json.loads(text) # {\'version\': 3, ...}\n\n# File round-trip\nwith open("config.json", "w", encoding="utf-8") as f:\n json.dump(config, f, indent=2)\n\nwith open("config.json", encoding="utf-8") as f:\n loaded = json.load(f)', ] def _pm_run_earlier(_srcs, _g): import sys as _s, io as _io, warnings as _w for _src in _srcs: _o, _e = _s.stdout, _s.stderr _s.stdout = _s.stderr = _io.StringIO() try: with _w.catch_warnings(): _w.simplefilter('ignore') exec(compile(_src, '<earlier block>', 'exec'), _g) except (Exception, SystemExit): pass finally: _s.stdout, _s.stderr = _o, _e if 'matplotlib.pyplot' in _s.modules: _s.modules['matplotlib.pyplot'].close('all') # their plots belong to their own blocks _pm_run_earlier(_pm_earlier, globals()) del _pm_earlier, _pm_run_earlier
Without sort_keys, Python keeps insertion order. That is fine in code, but in a file you diff, any change in key order shows up as a change, even though the data is the same.
5. JSON: Custom Types (datetime, Decimal, sets…)
json.dumps only knows about primitives: dict, list, str, int, float, bool, None. Hand it a datetime, Decimal, set, or any custom class and it raises TypeError: Object of type datetime is not JSON serializable.
There are three ways to handle them, from simplest to most complete:
(a) Convert before encoding. The simplest way, fine for one-off scripts:
from datetime import datetime record = {"name": "Alice", "joined": datetime.now().isoformat()} json.dumps(record)
setup added so this can run · re-runs 2 earlier blocks on this page first
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) json = _AutoMock('json') config = _AutoMock('config') # Earlier blocks on this page that this one builds on, re-run quietly first. try: import json from pathlib import Path except Exception: pass _pm_earlier = [ 'import json\nfrom pathlib import Path\n\nconfig = {"version": 3, "features": ["dark_mode", "autosave"], "limits": {"max": 100}}\n\n# String round-trip\ntext = json.dumps(config) # \'{"version": 3, ...}\'\nback = json.loads(text) # {\'version\': 3, ...}\n\n# File round-trip\nwith open("config.json", "w", encoding="utf-8") as f:\n json.dump(config, f, indent=2)\n\nwith open("config.json", encoding="utf-8") as f:\n loaded = json.load(f)', 'json.dumps(config, indent=2, sort_keys=True)', ] def _pm_run_earlier(_srcs, _g): import sys as _s, io as _io, warnings as _w for _src in _srcs: _o, _e = _s.stdout, _s.stderr _s.stdout = _s.stderr = _io.StringIO() try: with _w.catch_warnings(): _w.simplefilter('ignore') exec(compile(_src, '<earlier block>', 'exec'), _g) except (Exception, SystemExit): pass finally: _s.stdout, _s.stderr = _o, _e if 'matplotlib.pyplot' in _s.modules: _s.modules['matplotlib.pyplot'].close('all') # their plots belong to their own blocks _pm_run_earlier(_pm_earlier, globals()) del _pm_earlier, _pm_run_earlier
(b) default= callback. One line, for a quick custom encoder:
from datetime import datetime record = {"name": "Alice", "joined": datetime.now()} json.dumps(record, default=str) # falls back to str() for unknowns
setup added so this can run · re-runs 3 earlier blocks on this page first
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) json = _AutoMock('json') config = _AutoMock('config') # Earlier blocks on this page that this one builds on, re-run quietly first. try: import json from pathlib import Path from datetime import datetime except Exception: pass _pm_earlier = [ 'import json\nfrom pathlib import Path\n\nconfig = {"version": 3, "features": ["dark_mode", "autosave"], "limits": {"max": 100}}\n\n# String round-trip\ntext = json.dumps(config) # \'{"version": 3, ...}\'\nback = json.loads(text) # {\'version\': 3, ...}\n\n# File round-trip\nwith open("config.json", "w", encoding="utf-8") as f:\n json.dump(config, f, indent=2)\n\nwith open("config.json", encoding="utf-8") as f:\n loaded = json.load(f)', 'json.dumps(config, indent=2, sort_keys=True)', 'from datetime import datetime\nrecord = {"name": "Alice", "joined": datetime.now().isoformat()}\njson.dumps(record)', ] def _pm_run_earlier(_srcs, _g): import sys as _s, io as _io, warnings as _w for _src in _srcs: _o, _e = _s.stdout, _s.stderr _s.stdout = _s.stderr = _io.StringIO() try: with _w.catch_warnings(): _w.simplefilter('ignore') exec(compile(_src, '<earlier block>', 'exec'), _g) except (Exception, SystemExit): pass finally: _s.stdout, _s.stderr = _o, _e if 'matplotlib.pyplot' in _s.modules: _s.modules['matplotlib.pyplot'].close('all') # their plots belong to their own blocks _pm_run_earlier(_pm_earlier, globals()) del _pm_earlier, _pm_run_earlier
default=str works because str(datetime) returns an ISO-like string. For better control, supply your own function:
def encode(o): if isinstance(o, datetime): return o.isoformat() if isinstance(o, set): return sorted(o) raise TypeError(f"can't encode {type(o).__name__}") json.dumps(payload, default=encode)
setup added so this can run · defines payload, datetime, json
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) payload = _AutoMock('payload') datetime = _AutoMock('datetime') json = _AutoMock('json')
(c) Subclass json.JSONEncoder. Use this when you want the same encoding logic everywhere in a project:
class RichEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, datetime): return o.isoformat() if isinstance(o, set): return sorted(o) return super().default(o) json.dumps(payload, cls=RichEncoder, indent=2)
setup added so this can run · defines json, payload, datetime
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) json = _AutoMock('json') payload = _AutoMock('payload') datetime = _AutoMock('datetime')
The reverse direction, turning ISO strings back into datetime, uses object_hook:
from datetime import datetime import re ISO_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T") def parse_dates(obj): for k, v in obj.items(): if isinstance(v, str) and ISO_RE.match(v): try: obj[k] = datetime.fromisoformat(v) except ValueError: pass return obj json.loads('{"joined": "2026-05-14T10:00:00"}', object_hook=parse_dates) # {'joined': datetime.datetime(2026, 5, 14, 10, 0)}
setup added so this can run · defines json
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) json = _AutoMock('json')
object_hook is called for every JSON object as it's decoded. This is useful for typed parsing, but because it runs on every dict, keep it fast.
6. JSON Lines (JSONL / NDJSON): JSON for Streaming
In plain JSON, a collection is one big array, and you can't parse half of it. JSON Lines fixes that: one JSON object per line, no surrounding array, and you can add new records to the end of the file.
{"id": 1, "name": "Alice"}
{"id": 2, "name": "Bob"}
{"id": 3, "name": "Carol"}Writing JSONL: one json.dumps per record, then "\n":
import json from pathlib import Path def write_jsonl(path, records): with open(path, "w", encoding="utf-8") as f: for record in records: f.write(json.dumps(record, ensure_ascii=False)) f.write("\n")
Reading JSONL: one json.loads per line, as a generator so memory use stays constant:
def read_jsonl(path): with open(path, encoding="utf-8") as f: for line in f: line = line.strip() if line: # skip blank lines defensively yield json.loads(line) for record in read_jsonl("events.jsonl"): process(record)
setup added so this can run · defines process, json
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) def process(*_a, **_kw): print('-> process() called') return _AutoMock('process()') json = _AutoMock('json')
ensure_ascii=False is worth knowing. By default json.dumps escapes non-ASCII characters as \u sequences, making "café" become "caf\u00e9". That is still valid JSON, but much harder to read. Pass ensure_ascii=False and write UTF-8 throughout.
7. When to Use Which
| Use case | Format | Why |
|---|---|---|
| Tabular data, opens in spreadsheets | CSV | Universal, Excel-friendly, flat rows. |
| Nested config / single response | JSON | Native to most APIs, supports nesting. |
| Streaming logs, append-only event data, large datasets | JSONL | Line-by-line reads, easy to tail -f, no array wrapper. |
| Human-edited config files | TOML | Comments, sections, sane string handling. (Section 9) |
| Existing YAML ecosystem (Kubernetes, GitHub Actions) | YAML | Match the surrounding tools. (Section 9) |
Remember that JSON has no comments. People choose JSON as a config format and then find they can't add notes to it. Use TOML or YAML for human-edited config, and reserve JSON for machine-produced/consumed data.
For data that's both tabular and large (multi-GB CSV or JSONL), you must stream it. See Section 8.
8. Streaming Big Files: Don't Load Everything at Once
The biggest scaling mistake with both CSV and JSON is loading everything into memory. For huge files, stream row-by-row (CSV) or line-by-line (JSONL):
# CSV: DictReader already reads one row at a time. Don't turn it into a list. with open("huge.csv", encoding="utf-8", newline="") as f: for row in csv.DictReader(f): process(row) # constant memory # Bad: loads all 10 million rows into memory with open("huge.csv", encoding="utf-8", newline="") as f: rows = list(csv.DictReader(f)) # don't do this
setup added so this can run · defines csv, process
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) csv = _AutoMock('csv') def process(*_a, **_kw): print('-> process() called') return _AutoMock('process()')
For huge JSON (a single multi-GB array), the stdlib json module won't help, because it needs the whole array in memory. Use ijson (pip install ijson), which streams JSON one item at a time:
import ijson with open("huge.json", "rb") as f: for record in ijson.items(f, "item"): # one item of the top-level array at a time process(record)
setup added so this can run · defines process
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) def process(*_a, **_kw): print('-> process() called') return _AutoMock('process()')
Better yet, store large datasets as JSONL from the start. The stdlib can stream JSONL easily: one line, one json.loads. See generators for the iteration patterns that keep memory use constant.
9. TOML and YAML: For Config Files
Python 3.11 added tomllib to the stdlib for reading TOML (the format pyproject.toml uses). It can only read TOML. There is no tomllib.dumps. For writing, install tomli-w or tomlkit.
import tomllib from pathlib import Path with open("pyproject.toml", "rb") as f: # note: tomllib needs binary mode ("rb") config = tomllib.load(f) print(config["project"]["name"])
TOML is the modern default for Python project config. It is readable, typed and supports comments. Prefer it for new config files unless the surrounding ecosystem (e.g. Kubernetes manifests) is already YAML.
YAML isn't in the stdlib, so install it with pip install pyyaml:
import yaml with open("compose.yaml", encoding="utf-8") as f: config = yaml.safe_load(f) # ALWAYS safe_load, never plain load
Use yaml.safe_load, never yaml.load. Depending on the loader, yaml.load can create arbitrary Python objects from untrusted YAML, which can lead to code execution. It is the YAML equivalent of pickle.loads on untrusted data, and it has caused real security vulnerabilities. safe_load is the only safe default.
10. A Short Note on pandas
For data-science workflows on tabular data, pandas.read_csv and pandas.read_json are the right tools. They handle dialects, types, missing values and gigabytes of data in one line:
import pandas as pd df = pd.read_csv("huge.csv", chunksize=10_000) # streams in chunks for chunk in df: process(chunk)
setup added so this can run · defines process
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) def process(*_a, **_kw): print('-> process() called') return _AutoMock('process()')
For one-off CSV manipulation or anything that fits in RAM, pandas is faster to write, more expressive, and worth the import. For long-running services where the number of dependencies matters, or when you really need to process one line at a time, the stdlib csv module is the right choice.
This is a Python how-to, not a data-science course. The data science lessons on this site cover pandas properly.
11. Common Mistakes
1. Forgetting newline="" on CSV files
Blank rows on Windows. Always pass newline="" to both open() for read and write.
2. Defaulting the encoding
On Windows, open() often defaults to a legacy encoding such as cp1252, not UTF-8. Non-ASCII characters then come out garbled ("mojibake") or raise an error. Always pass encoding="utf-8" (or "utf-8-sig" for Excel-exported files). See fileio.
3. json.dumps(datetime.now()) → TypeErrorjson doesn't know about datetime, Decimal, set, or your custom classes. Use default=str for quick fixes, and a custom default= callable or a JSONEncoder subclass for production.
4. Using JSON as a config format
JSON has no comments, and people who edit config files want to add notes. Use TOML (or YAML) for human-edited config, and reserve JSON for machine-produced data.
5. list(reader) on huge CSVs
Pulls everything into memory. csv.DictReader is already an iterator, so use for row in reader: and process one row at a time. The same goes for JSONL: process each record as you read it instead of collecting them.
6. Loading multi-GB JSON arrays with json.load
The stdlib can't stream JSON. If you must read huge JSON, use ijson. Better: produce JSONL instead, where streaming is one line at a time.
7. Hand-writing CSV with ",".join(row)
Fields containing commas, quotes, or newlines will break your file. Use csv.writer, which escapes them correctly.
8. yaml.load on untrusted YAML
It is a code execution risk. Always use yaml.safe_load.
🎯 Your Turn: Stream a CSV into JSONL
Write convert_csv_to_jsonl(csv_path, jsonl_path) that reads a CSV row by row and writes one JSON object per line to a JSONL file. Constraints:
- O(1) memory: it must handle multi-GB inputs without loading them into memory.
- UTF-8 throughout. Handle input that starts with a BOM without any extra code.
- Use the header row as field names (
csv.DictReaderdoes this for you). ensure_ascii=Falsein the output so non-ASCII characters stay human-readable.
Example:
# input.csv name,age,city Alice,30,Bengaluru Bob,25,München
setup added so this can run · defines name, age, city, Alice, Bengaluru, Bob, München
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) name = _AutoMock('name') age = _AutoMock('age') city = _AutoMock('city') Alice = _AutoMock('Alice') Bengaluru = _AutoMock('Bengaluru') Bob = _AutoMock('Bob') München = _AutoMock('München')
# output.jsonl {"name": "Alice", "age": "30", "city": "Bengaluru"} {"name": "Bob", "age": "25", "city": "München"}
Skeleton:
import csv import json from pathlib import Path def convert_csv_to_jsonl(csv_path, jsonl_path): # TODO 1: open csv with utf-8-sig + newline="" # TODO 2: open jsonl with utf-8 for writing # TODO 3: DictReader on the input # TODO 4: for each row, json.dumps + "\n" to the output ...
Hint 1: Why utf-8-sig for input
Excel-exported CSVs often start with a UTF-8 BOM (\ufeff). With plain utf-8, that BOM becomes the first character of your first column header, and row["name"] raises KeyError because the real key is "\ufeffname". utf-8-sig removes it automatically.
Hint 2: Streaming write
Don't collect rows into a list first.csv.DictReader is an iterator, so loop over it directly and write each row immediately. Memory stays at one row, no matter how big the file is.
Show full solution
import csv import json from pathlib import Path def convert_csv_to_jsonl(csv_path, jsonl_path): """Stream a CSV file row-by-row into a JSONL file. O(1) memory.""" csv_path = Path(csv_path) jsonl_path = Path(jsonl_path) with csv_path.open("r", encoding="utf-8-sig", newline="") as src, \ jsonl_path.open("w", encoding="utf-8") as dst: reader = csv.DictReader(src) for row in reader: dst.write(json.dumps(row, ensure_ascii=False)) dst.write("\n") # Demo import tempfile from pathlib import Path with tempfile.TemporaryDirectory() as tmpdir: tmp = Path(tmpdir) src = tmp / "input.csv" dst = tmp / "output.jsonl" src.write_text( "name,age,city\n" "Alice,30,Bengaluru\n" "Bob,25,München\n", encoding="utf-8", ) convert_csv_to_jsonl(src, dst) print(dst.read_text(encoding="utf-8")) # {"name": "Alice", "age": "30", "city": "Bengaluru"} # {"name": "Bob", "age": "25", "city": "München"}
What's working here:
utf-8-sigon input strips the BOM if present, leaving keys clean ("name", not"\ufeffname").newline=""lets thecsvmodule handle newlines itself, so there are no blank rows on Windows.csv.DictReaderuses the header row as keys automatically. No hand-built schema needed.- Iteration, not a list:
for row in reader:reads one row at a time. Memory use is the same for a 10-MB file and a 10-GB file. ensure_ascii=FalsekeepsMünchenasMünchenin the output, notM\u00fcnchen.- Two files in one
with: both close correctly, even if writing raises an error halfway through.
Notes you'd want for production use:
- All values are strings in the output, because CSV has no type information. If you need real types (
"age": 30instead of"age": "30"), add a type-conversion step, usually a per-column mapping like{"age": int, "joined": datetime.fromisoformat}. - No header? No DictReader. Use
csv.readerand build dicts yourself with a known field list. - Errors aren't caught: a malformed row will stop the conversion halfway. Wrap the inner write in
try/except csv.Errorif you want a forgiving converter that logs and skips bad rows. See exceptions.
This structure (open both files, loop, transform, write one line at a time) is the standard pattern for streaming conversions. The same structure works for JSONL→CSV, CSV→Parquet and log→event-stream.
What You Learned
- CSV:
csv.DictReader/DictWriter, always withnewline=""andencoding="utf-8"(orutf-8-sigfor BOM-prefixed inputs). - CSV dialects: comma/tab/pipe,
QUOTE_ALL/QUOTE_NONNUMERIC,csv.Snifferfor unknown formats. - JSON: four functions,
dumps/loads/dump/load.indent=2for humans,sort_keys=Truefor diff-friendly output,ensure_ascii=Falsefor UTF-8 readability. - Custom types in JSON:
default=strfor quick fixes,default=callablefor control,JSONEncodersubclass for project-wide encoders. object_hookparses typed values back out on decode (ISO strings →datetime).- JSONL: one JSON object per line. The right format for streaming logs, append-only event data, and anything larger than RAM.
- Format choice: CSV for tabular spreadsheet-friendly, JSON for nested API payloads, JSONL for streams, TOML/YAML for human config.
- Stream, and don't load everything at once.
for row in DictReader(f):andfor line in f:stay O(1). Avoidlist(reader)andjson.loadon multi-GB files. - TOML is stdlib in 3.11+ (read-only). YAML needs
pyyaml. Always usesafe_load, neverload. - pandas for data-science workflows, stdlib for streaming and lean production code.
Next: Automation Scripts. Combine these formats into small utilities you can run every day.
Practice this
on practicepython.inShort exercises that run in your browser and tell you what your code did, as well as whether a test passed.
Found this useful? Share it with a friend who’s learning Python.