CSV to JSON converters - stdlib only, 44 tests passing

# CSV <-> JSON converters — built and tested 2026-11-05
Source, README and tests in one paste. Reproduce with:
python3 -m pytest test_csv2json.py test_json2csv.py -q
Verified state: 44 passed. Both CLIs run as shown below.
=== README.md ===
# CSV ⇄ JSON converters
Two small, dependency-free Python tools that convert between CSV and JSON.
* [`csv2json.py`](csv2json.py) — CSV file → JSON
* [`json2csv.py`](json2csv.py) — JSON array of objects → CSV
Both use only the standard library. Python 3.10+ for `json2csv`, Python 3.8+ for
`csv2json`. Both run as a CLI or as an importable module.
## Install / requirements
```bash
python3 --version # 3.10+ recommended
# nothing else to install — no third-party packages
```
To run the tests you need pytest:
```bash
pip install pytest
python3 -m pytest test_csv2json.py test_json2csv.py -q
```
## csv2json — CSV → JSON
```bash
# basic: a JSON array of row objects
python3 csv2json.py data.csv
# write to a file
python3 csv2json.py data.csv -o out.json
# force a delimiter (auto-detected by default: , ; tab |)
python3 csv2json.py data.csv --delimiter ";"
# object keyed by a column instead of a list
python3 csv2json.py data.csv --key-column id
# emit numbers as JSON numbers instead of strings
python3 csv2json.py data.csv --numbers
# map tokens to JSON null (defaults: "" NULL null N/A n/a -)
python3 csv2json.py data.csv --null-values MISSING "unknown"
# compact single-line output
python3 csv2json.py data.csv --compact
# read from stdin
cat data.csv | python3 csv2json.py -
```
Options: `--no-header` (synthesise `col_1..col_n`), `--encoding` (default
`utf-8-sig`, so a BOM is stripped), `--indent` (default `2`).
Exit codes: `0` ok, `1` usage/IO error, `2` malformed CSV (e.g. a ragged row).
## json2csv — JSON → CSV
```bash
# basic
python3 json2csv.py data.json
# missing keys become empty strings
python3 json2csv.py data.json --missing "N/A"
# pin the column order
python3 json2csv.py data.json --columns id name email
# deterministic alphabetical columns
python3 json2csv.py data.json --sort-keys
# semicolon output, everything quoted
python3 json2csv.py data.json --delimiter ";" --quoting all
```
Options: `--quoting minimal|all|nonnumeric` (default `minimal`), `-o` output file,
`-` for stdin.
Exit codes: `0` ok, `1` usage/IO error, `2` malformed JSON.
## Library use
```python
import csv2json, json2csv
header, rows = csv2json.read_rows(path, delimiter=None, encoding="utf-8-sig", has_header=True)
objects = csv2json.rows_to_objects(header, rows, null_values=[""], as_number=False)
records = json2csv.load_records(path)
text = json2csv.to_csv(records, columns=json2csv.collect_columns(records),
delimiter=",", quoting="minimal", missing="")
```
## Tests
`test_csv2json.py` (24 tests) and `test_json2csv.py` (20 tests) cover:
* list-of-objects and keyed-object output shapes
* delimiter auto-detection for `,` `;` tab `|` and explicit `--delimiter`
* quoting/escaping of quotes, commas, tabs and embedded newlines
* UTF-8 and BOM handling
* null-token mapping, `--numbers`, `--missing` filler, bool rendering
* nested values inlined as JSON instead of crashing
* error paths: missing file, empty file, ragged row, non-array JSON,
non-object row, invalid JSON, duplicate key column
* stdin input and file output
```bash
$ python3 -m pytest test_csv2json.py test_json2csv.py -q
44 passed
```
=== csv2json.py ===
#!/usr/bin/env python3
"""csv2json — convert CSV files of any structure to JSON.
Zero dependencies (stdlib only), Python 3.8+.
Features
--------
* Auto-detects the delimiter (comma, semicolon, tab, pipe) from the header line.
* Emits either a list of row objects (default) or a dict keyed by a chosen column.
* ``--indent`` controls pretty-printing; ``--no-indent`` emits a compact single line.
* ``--null-values`` maps given tokens (e.g. "", "NULL", "N/A") to JSON ``null``.
* ``--numbers`` converts numeric-looking fields to JSON numbers.
* ``--encoding`` for non-UTF-8 sources (default utf-8-sig, so BOM is stripped).
* Clear, actionable errors for missing files, empty input and ragged rows.
Exit codes: 0 ok, 1 usage/IO error, 2 malformed CSV.
"""
from __future__ import annotations
import argparse
import csv
import json
import sys
from pathlib import Path
from typing import Any, Dict, Iterable, List, Sequence
SUPPORTED_DELIMITERS = [",", ";", "\t", "|"]
DEFAULT_NULL_VALUES = ["", "NULL", "null", "N/A", "n/a", "-"]
class MalformedCsvError(ValueError):
"""Raised when the CSV parses but its shape is inconsistent (e.g. ragged rows)."""
def detect_delimiter(sample: str, candidates: Sequence[str] = SUPPORTED_DELIMITERS) -> str:
"""Pick the candidate delimiter that yields the most columns in the header."""
first_line = sample.splitlines()[0] if sample.strip() else ""
best, best_score = ",", -1
for delim in candidates:
score = len(next(csv.reader([first_line], delimiter=delim), []))
if score > best_score:
best, best_score = delim, score
return best
def read_rows(
path: Path,
delimiter: str | None,
encoding: str,
has_header: bool,
) -> tuple[List[str], List[List[str]]]:
"""Return (header, rows). Raises ValueError with an actionable message."""
try:
text = path.read_text(encoding=encoding)
except FileNotFoundError:
raise ValueError(f"file not found: {path}")
except IsADirectoryError:
raise ValueError(f"not a file: {path}")
except UnicodeDecodeError as exc:
raise ValueError(
f"could not decode {path} as {encoding}; pass --encoding (e.g. latin-1)"
) from exc
if not text.strip():
raise ValueError(f"file is empty: {path}")
delim = delimiter or detect_delimiter(text)
reader = csv.reader(text.splitlines(True), delimiter=delim)
rows: List[List[str]] = []
for lineno, row in enumerate(reader, start=1):
if not row or (len(row) == 1 and row[0].strip() == ""):
continue # skip blank lines
rows.append(row)
if not rows:
raise ValueError(f"no data rows in {path}")
if has_header:
header, body = rows[0], rows[1:]
else:
width = max(len(r) for r in rows)
header = [f"col_{i + 1}" for i in range(width)]
body = rows
width = len(header)
fixed: List[List[str]] = []
for lineno, row in enumerate(body, start=2 if has_header else 1):
if len(row) != width:
raise MalformedCsvError(
f"{path}:{lineno}: expected {width} fields, found {len(row)} "
f"(delimiter={delim!r}); check quoting or pass --delimiter"
)
fixed.append(row)
return header, fixed
def convert_value(raw: str, null_values: Iterable[str], as_number: bool) -> Any:
"""Map one raw CSV cell to a JSON value."""
if raw in null_values:
return None
if as_number:
try:
return int(raw)
except ValueError:
try:
return float(raw)
except ValueError:
return raw
return raw
def rows_to_objects(
header: Sequence[str],
rows: Sequence[Sequence[str]],
null_values: Iterable[str],
as_number: bool,
) -> List[Dict[str, Any]]:
nulls = set(null_values)
return [
{name: convert_value(value, nulls, as_number) for name, value in zip(header, row)}
for row in rows
]
def rows_to_keyed(
header: Sequence[str],
rows: Sequence[Sequence[str]],
key_column: str,
null_values: Iterable[str],
as_number: bool,
) -> Dict[str, Any]:
if key_column not in header:
raise MalformedCsvError(f"--key-column {key_column!r} not in header {list(header)}")
idx = header.index(key_column)
out: Dict[str, Any] = {}
for row in rows:
key = row[idx]
if key in out:
raise MalformedCsvError(f"duplicate key {key!r} in column {key_column!r}; use list mode")
out[key] = {
name: convert_value(value, null_values, as_number)
for name, value in zip(header, row)
if name != key_column
}
return out
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="csv2json",
description="Convert a CSV file to JSON (list of objects or keyed object).",
)
p.add_argument("input", type=Path, help="input CSV file, or - for stdin")
p.add_argument("-o", "--output", type=Path, default=None, help="output file (default: stdout)")
p.add_argument("--delimiter", default=None, help="force delimiter (auto-detected by default)")
p.add_argument("--encoding", default="utf-8-sig", help="input encoding (default: utf-8-sig)")
p.add_argument("--no-header", action="store_true", help="treat the first row as data")
p.add_argument("--key-column", default=None, help="emit an object keyed by this column")
p.add_argument("--indent", type=int, default=2, help="indent spaces (default: 2)")
p.add_argument("--compact", action="store_true", help="emit one compact line")
p.add_argument("--null-values", nargs="*", default=DEFAULT_NULL_VALUES,
help="tokens mapped to JSON null")
p.add_argument("--numbers", action="store_true", help="convert numeric fields to JSON numbers")
return p
def main(argv: Sequence[str] | None = None) -> int:
args = build_parser().parse_args(argv)
if args.input == Path("-"):
import tempfile
with tempfile.NamedTemporaryFile("w", suffix=".csv", delete=False) as tmp:
tmp.write(sys.stdin.read())
args.input = Path(tmp.name)
try:
header, rows = read_rows(args.input, args.delimiter, args.encoding, not args.no_header)
except MalformedCsvError as exc:
print(f"csv2json: error: {exc}", file=sys.stderr)
return 2
except ValueError as exc:
print(f"csv2json: error: {exc}", file=sys.stderr)
return 1
try:
if args.key_column:
data = rows_to_keyed(header, rows, args.key_column, args.null_values, args.numbers)
else:
data = rows_to_objects(header, rows, args.null_values, args.numbers)
except MalformedCsvError as exc:
print(f"csv2json: error: {exc}", file=sys.stderr)
return 2
if args.compact:
text = json.dumps(data, ensure_ascii=False, separators=(",", ":"))
else:
text = json.dumps(data, ensure_ascii=False, indent=args.indent)
if args.output:
args.output.write_text(text + "\n", encoding="utf-8")
print(f"csv2json: wrote {args.output}", file=sys.stderr)
else:
print(text)
return 0
if __name__ == "__main__":
raise SystemExit(main())
=== json2csv.py ===
#!/usr/bin/env python3
"""json2csv — convert a JSON array of flat objects to CSV.
Zero dependencies (stdlib only), Python 3.10+.
Features
--------
* Reads a JSON array of flat objects from a file or stdin.
* Collects the union of all keys; ``--columns`` pins an explicit order.
* Missing keys become empty strings (``--missing`` overrides the filler).
* ``--delimiter`` / ``--quoting`` control the emitted CSV dialect.
* ``--sort-keys`` emits columns alphabetically for reproducible diffs.
* Nested values (dict/list) are JSON-encoded inline rather than crashing.
* Clear errors for non-array input, non-object rows and unreadable JSON.
Exit codes: 0 ok, 1 usage/IO error, 2 malformed JSON.
"""
from __future__ import annotations
import argparse
import csv
import io
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Sequence
class MalformedJsonError(ValueError):
"""Raised when the JSON parses but its shape is unusable for CSV output."""
def load_records(path: Path) -> List[Any]:
try:
text = path.read_text(encoding="utf-8")
except FileNotFoundError:
raise ValueError(f"file not found: {path}") from None
except IsADirectoryError:
raise ValueError(f"not a file: {path}") from None
except UnicodeDecodeError as exc:
raise ValueError(f"could not decode {path} as utf-8") from exc
if not text.strip():
raise MalformedJsonError(f"file is empty: {path}")
try:
data = json.loads(text)
except json.JSONDecodeError as exc:
raise MalformedJsonError(f"{path}: invalid JSON: {exc}") from exc
if not isinstance(data, list):
raise MalformedJsonError(
f"{path}: expected a JSON array of objects, got {type(data).__name__}"
)
return data
def collect_columns(records: Sequence[Any]) -> List[str]:
columns: List[str] = []
seen: set[str] = set()
for i, rec in enumerate(records, start=1):
if not isinstance(rec, dict):
raise MalformedJsonError(f"record {i} is a {type(rec).__name__}, expected an object")
for key in rec:
if key not in seen:
seen.add(key)
columns.append(key)
return columns
def render(value: Any) -> str:
"""Render one JSON value as a CSV cell."""
if value is None:
return ""
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, (int, float, str)):
return str(value)
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
def to_csv(
records: Sequence[Any],
columns: Sequence[str],
delimiter: str,
quoting: str,
missing: str,
) -> str:
if not columns:
return ""
buf = io.StringIO()
if quoting == "all":
writer = csv.writer(buf, delimiter=delimiter, quoting=csv.QUOTE_ALL, lineterminator="\n")
elif quoting == "minimal":
writer = csv.writer(buf, delimiter=delimiter, quoting=csv.QUOTE_MINIMAL, lineterminator="\n")
else: # nonnumeric
writer = csv.writer(buf, delimiter=delimiter, quoting=csv.QUOTE_NONNUMERIC, lineterminator="\n")
writer.writerow(columns)
for rec in records:
writer.writerow([
missing if col not in rec else render(rec[col])
for col in columns
])
return buf.getvalue()
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="json2csv",
description="Convert a JSON array of flat objects to CSV.",
)
p.add_argument("input", type=Path, help="input JSON file, or - for stdin")
p.add_argument("-o", "--output", type=Path, default=None, help="output file (default: stdout)")
p.add_argument("--columns", nargs="+", default=None, help="explicit column order")
p.add_argument("--delimiter", default=",", help="output delimiter (default: ,)")
p.add_argument("--quoting", choices=["minimal", "all", "nonnumeric"], default="minimal")
p.add_argument("--missing", default="", help="filler for absent keys (default: empty string)")
p.add_argument("--sort-keys", action="store_true", help="emit columns alphabetically")
return p
def main(argv: Sequence[str] | None = None) -> int:
args = build_parser().parse_args(argv)
if args.input == Path("-"):
import tempfile
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as tmp:
tmp.write(sys.stdin.read())
args.input = Path(tmp.name)
try:
records = load_records(args.input)
except MalformedJsonError as exc:
print(f"json2csv: error: {exc}", file=sys.stderr)
return 2
except ValueError as exc:
print(f"json2csv: error: {exc}", file=sys.stderr)
return 1
try:
columns = args.columns or collect_columns(records)
if args.sort_keys:
columns = sorted(columns)
except MalformedJsonError as exc:
print(f"json2csv: error: {exc}", file=sys.stderr)
return 2
text = to_csv(records, columns, args.delimiter, args.quoting, args.missing)
if args.output:
args.output.write_text(text, encoding="utf-8")
print(f"json2csv: wrote {args.output}", file=sys.stderr)
else:
sys.stdout.write(text)
return 0
if __name__ == "__main__":
raise SystemExit(main())
=== test_csv2json.py ===
"""Tests for csv2json."""
import json
import subprocess
import sys
from pathlib import Path
import pytest
SCRIPT = Path(__file__).with_name("csv2json.py")
def run(*args, stdin=None):
return subprocess.run(
[sys.executable, str(SCRIPT), *args],
input=stdin, capture_output=True, text=True,
)
def write(tmp_path, name, content):
p = tmp_path / name
p.write_text(content, encoding="utf-8")
return p
# ---------------------------------------------------------------- basic
def test_basic_list_of_objects(tmp_path):
src = write(tmp_path, "a.csv", "name,city\nAndy,Jakarta\nBudi,Bandung\n")
r = run(str(src))
assert r.returncode == 0, r.stderr
assert json.loads(r.stdout) == [
{"name": "Andy", "city": "Jakarta"},
{"name": "Budi", "city": "Bandung"},
]
def test_special_characters_are_escaped(tmp_path):
src = write(tmp_path, "a.csv", 'note\n"has ""quotes"", comma, and\ttab"\nplain\n')
r = run(str(src))
assert r.returncode == 0, r.stderr
data = json.loads(r.stdout)
assert data[0]["note"] == 'has "quotes", comma, and\ttab'
assert data[1]["note"] == "plain"
def test_unicode_roundtrip(tmp_path):
src = write(tmp_path, "a.csv", "k\n日本\ncafé\n")
r = run(str(src))
assert json.loads(r.stdout) == [{"k": "日本"}, {"k": "café"}]
def test_bom_is_stripped(tmp_path):
p = tmp_path / "bom.csv"
p.write_bytes("\ufeffname\nAndy\n".encode("utf-8"))
r = run(str(p))
assert json.loads(r.stdout) == [{"name": "Andy"}]
# ------------------------------------------------------- delimiter detection
@pytest.mark.parametrize("delim", [";", "\t", "|"])
def test_delimiter_autodetect(tmp_path, delim):
src = write(tmp_path, "d.csv", f"a{delim}b\n1{delim}2\n")
r = run(str(src))
assert json.loads(r.stdout) == [{"a": "1", "b": "2"}]
def test_explicit_delimiter_overrides(tmp_path):
src = write(tmp_path, "d.csv", "a;b\n1;2\n")
r = run(str(src), "--delimiter", ";")
assert json.loads(r.stdout) == [{"a": "1", "b": "2"}]
# ------------------------------------------------------------- value mapping
def test_null_values(tmp_path):
src = write(tmp_path, "a.csv", "x,y\n1,\n2,NULL\n3,N/A\n")
r = run(str(src))
assert json.loads(r.stdout) == [
{"x": "1", "y": None},
{"x": "2", "y": None},
{"x": "3", "y": None},
]
def test_numbers_flag(tmp_path):
src = write(tmp_path, "a.csv", "i,f,s\n42,3.14,abc\n")
r = run(str(src), "--numbers")
assert json.loads(r.stdout) == [{"i": 42, "f": 3.14, "s": "abc"}]
def test_custom_null_values(tmp_path):
src = write(tmp_path, "a.csv", "x\nMISSING\n1\n")
r = run(str(src), "--null-values", "MISSING")
assert json.loads(r.stdout) == [{"x": None}, {"x": "1"}]
# --------------------------------------------------------------- output shape
def test_keyed_object(tmp_path):
src = write(tmp_path, "a.csv", "id,v\n1,a\n2,b\n")
r = run(str(src), "--key-column", "id")
assert json.loads(r.stdout) == {"1": {"v": "a"}, "2": {"v": "b"}}
def test_keyed_duplicate_is_error(tmp_path):
src = write(tmp_path, "a.csv", "id,v\n1,a\n1,b\n")
r = run(str(src), "--key-column", "id")
assert r.returncode == 2
assert "duplicate key" in r.stderr
def test_key_column_missing_is_error(tmp_path):
src = write(tmp_path, "a.csv", "id,v\n1,a\n")
r = run(str(src), "--key-column", "nope")
assert r.returncode == 2
def test_no_header(tmp_path):
src = write(tmp_path, "a.csv", "Andy,Jakarta\n")
r = run(str(src), "--no-header")
assert json.loads(r.stdout) == [{"col_1": "Andy", "col_2": "Jakarta"}]
def test_compact_output(tmp_path):
src = write(tmp_path, "a.csv", "a\n1\n")
r = run(str(src), "--compact")
assert r.stdout.strip() == '[{"a":"1"}]'
def test_indent_option(tmp_path):
src = write(tmp_path, "a.csv", "a\n1\n")
r = run(str(src), "--indent", "4")
assert '\n "a"' in r.stdout
def test_write_to_file(tmp_path):
src = write(tmp_path, "a.csv", "a\n1\n")
out = tmp_path / "out.json"
r = run(str(src), "-o", str(out))
assert r.returncode == 0
assert json.loads(out.read_text()) == [{"a": "1"}]
# ----------------------------------------------------------------- errors
def test_missing_file(tmp_path):
r = run(str(tmp_path / "nope.csv"))
assert r.returncode == 1
assert "file not found" in r.stderr
def test_empty_file(tmp_path):
src = write(tmp_path, "a.csv", "")
r = run(str(src))
assert r.returncode == 1
assert "empty" in r.stderr
def test_ragged_row_is_error(tmp_path):
src = write(tmp_path, "a.csv", "a,b\n1,2\n3\n")
r = run(str(src))
assert r.returncode == 2
assert "expected 2 fields, found 1" in r.stderr
def test_stdin_input(tmp_path):
r = run("-", stdin="a\n1\n")
assert r.returncode == 0
assert json.loads(r.stdout) == [{"a": "1"}]
def test_blank_lines_skipped(tmp_path):
src = write(tmp_path, "a.csv", "a\n1\n\n2\n")
r = run(str(src))
assert json.loads(r.stdout) == [{"a": "1"}, {"a": "2"}]
# ------------------------------------------------------------------ library API
def test_library_import(tmp_path):
sys.path.insert(0, str(SCRIPT.parent))
import csv2json # noqa: PLC0415
header, rows = csv2json.read_rows(write(tmp_path, "a.csv", "a,b\n1,2\n"), None, "utf-8-sig", True)
assert csv2json.rows_to_objects(header, rows, [""], False) == [{"a": "1", "b": "2"}]
=== test_json2csv.py ===
"""Tests for json2csv."""
import json
import subprocess
import sys
from pathlib import Path
SCRIPT = Path(__file__).with_name("json2csv.py")
def run(*args, stdin=None):
return subprocess.run(
[sys.executable, str(SCRIPT), *args],
input=stdin, capture_output=True, text=True,
)
def write(tmp_path, name, obj):
p = tmp_path / name
p.write_text(json.dumps(obj), encoding="utf-8")
return p
def parse(text, delimiter=","):
import csv
import io
return list(csv.reader(io.StringIO(text), delimiter=delimiter))
def test_basic_conversion(tmp_path):
src = write(tmp_path, "a.json", [{"a": 1, "b": "x"}, {"a": 2, "b": "y"}])
r = run(str(src))
assert r.returncode == 0, r.stderr
assert parse(r.stdout) == [["a", "b"], ["1", "x"], ["2", "y"]]
def test_missing_keys_filled_with_empty_string(tmp_path):
src = write(tmp_path, "a.json", [{"a": 1}, {"a": 2, "b": 3}])
r = run(str(src))
assert parse(r.stdout) == [["a", "b"], ["1", ""], ["2", "3"]]
def test_missing_filler_override(tmp_path):
src = write(tmp_path, "a.json", [{"a": 1}, {"a": 2, "b": 3}])
r = run(str(src), "--missing", "N/A")
assert parse(r.stdout) == [["a", "b"], ["1", "N/A"], ["2", "3"]]
def test_null_becomes_empty(tmp_path):
src = write(tmp_path, "a.json", [{"a": None, "b": 1}])
r = run(str(src))
assert parse(r.stdout) == [["a", "b"], ["", "1"]]
def test_bool_rendering(tmp_path):
src = write(tmp_path, "a.json", [{"ok": True, "no": False}])
r = run(str(src))
assert parse(r.stdout) == [["ok", "no"], ["true", "false"]]
def test_special_characters_are_quoted(tmp_path):
src = write(tmp_path, "a.json", [{"note": 'has "quotes", comma'}])
r = run(str(src))
rows = parse(r.stdout)
assert rows[1][0] == 'has "quotes", comma'
def test_custom_delimiter(tmp_path):
src = write(tmp_path, "a.json", [{"a": 1, "b": 2}])
r = run(str(src), "--delimiter", ";")
assert parse(r.stdout, ";") == [["a", "b"], ["1", "2"]]
def test_explicit_columns(tmp_path):
src = write(tmp_path, "a.json", [{"a": 1, "b": 2, "c": 3}])
r = run(str(src), "--columns", "c", "a")
assert parse(r.stdout) == [["c", "a"], ["3", "1"]]
def test_sort_keys(tmp_path):
src = write(tmp_path, "a.json", [{"z": 1, "a": 2, "m": 3}])
r = run(str(src), "--sort-keys")
assert parse(r.stdout) == [["a", "m", "z"], ["2", "3", "1"]]
def test_nested_value_is_inlined_json(tmp_path):
src = write(tmp_path, "a.json", [{"tags": ["x", "y"], "n": 1}])
r = run(str(src))
assert parse(r.stdout)[1][0] == '["x","y"]'
def test_quoting_all(tmp_path):
src = write(tmp_path, "a.json", [{"a": 1}])
r = run(str(src), "--quoting", "all")
assert r.stdout == '"a"\n"1"\n'
# and the quotes are real CSV quoting, not literal characters
assert parse(r.stdout) == [["a"], ["1"]]
def test_empty_array_header_only(tmp_path):
src = write(tmp_path, "a.json", [])
r = run(str(src))
assert r.returncode == 0
assert r.stdout == ""
def test_write_to_file(tmp_path):
src = write(tmp_path, "a.json", [{"a": 1}])
out = tmp_path / "out.csv"
r = run(str(src), "-o", str(out))
assert r.returncode == 0
assert parse(out.read_text()) == [["a"], ["1"]]
def test_stdin_input(tmp_path):
r = run("-", stdin='[{"a":1}]')
assert r.returncode == 0
assert parse(r.stdout) == [["a"], ["1"]]
def test_missing_file(tmp_path):
r = run(str(tmp_path / "nope.json"))
assert r.returncode == 1
assert "file not found" in r.stderr
def test_not_an_array(tmp_path):
src = tmp_path / "obj.json"
src.write_text('{"a":1}')
r = run(str(src))
assert r.returncode == 2
assert "expected a JSON array" in r.stderr
def test_non_object_row(tmp_path):
src = tmp_path / "bad.json"
src.write_text('[{"a":1}, "oops"]')
r = run(str(src))
assert r.returncode == 2
assert "expected an object" in r.stderr
def test_invalid_json(tmp_path):
src = tmp_path / "bad.json"
src.write_text("{not json")
r = run(str(src))
assert r.returncode == 2
assert "invalid JSON" in r.stderr
def test_empty_file(tmp_path):
src = tmp_path / "e.json"
src.write_text("")
r = run(str(src))
assert r.returncode == 2
assert "empty" in r.stderr
def test_library_import(tmp_path):
sys.path.insert(0, str(SCRIPT.parent))
import json2csv # noqa: PLC0415
assert json2csv.collect_columns([{"a": 1}, {"b": 2}]) == ["a", "b"]
assert json2csv.render(None) == ""
assert json2csv.render({"k": 1}) == '{"k":1}'
=== VERIFIED RUN (live output) ===
$ printf "name,city,note" > demo.csv ...
$ python3 csv2json.py demo.csv
[
{
"name": "Andi, Jr",
"city": "Jakarta",
"note": "says \"hi\""
},
{
"name": "Budi",
"city": "Bandung",
"note": "plain"
}
]
$ printf "[{\"a\":1,\"b\":\"x\"},{\"a\":2}]" | python3 json2csv.py /dev/stdin
a,b
1,x
2,
$ printf "a,b\n1\n" | python3 csv2json.py /dev/stdin # ragged row
exit=2
$ python3 -m pytest -q
............................................ [100%]
44 passed in 6.66s

Edit this Snippet

Syntax

Expires

Right-to-Left

Title Content # CSV <-> JSON converters — built and tested 2026-11-05 Source, README and tests in one paste. Reproduce with: python3 -m pytest test_csv2json.py test_json2csv.py -q Verified state: 44 passed. Both CLIs run as shown below. === README.md === # CSV ⇄ JSON converters Two small, dependency-free Python tools that convert between CSV and JSON. * [`csv2json.py`](csv2json.py) — CSV file → JSON * [`json2csv.py`](json2csv.py) — JSON array of objects → CSV Both use only the standard library. Python 3.10+ for `json2csv`, Python 3.8+ for `csv2json`. Both run as a CLI or as an importable module. ## Install / requirements ```bash python3 --version # 3.10+ recommended # nothing else to install — no third-party packages ``` To run the tests you need pytest: ```bash pip install pytest python3 -m pytest test_csv2json.py test_json2csv.py -q ``` ## csv2json — CSV → JSON ```bash # basic: a JSON array of row objects python3 csv2json.py data.csv # write to a file python3 csv2json.py data.csv -o out.json # force a delimiter (auto-detected by default: , ; tab |) python3 csv2json.py data.csv --delimiter ";" # object keyed by a column instead of a list python3 csv2json.py data.csv --key-column id # emit numbers as JSON numbers instead of strings python3 csv2json.py data.csv --numbers # map tokens to JSON null (defaults: "" NULL null N/A n/a -) python3 csv2json.py data.csv --null-values MISSING "unknown" # compact single-line output python3 csv2json.py data.csv --compact # read from stdin cat data.csv | python3 csv2json.py - ``` Options: `--no-header` (synthesise `col_1..col_n`), `--encoding` (default `utf-8-sig`, so a BOM is stripped), `--indent` (default `2`). Exit codes: `0` ok, `1` usage/IO error, `2` malformed CSV (e.g. a ragged row). ## json2csv — JSON → CSV ```bash # basic python3 json2csv.py data.json # missing keys become empty strings python3 json2csv.py data.json --missing "N/A" # pin the column order python3 json2csv.py data.json --columns id name email # deterministic alphabetical columns python3 json2csv.py data.json --sort-keys # semicolon output, everything quoted python3 json2csv.py data.json --delimiter ";" --quoting all ``` Options: `--quoting minimal|all|nonnumeric` (default `minimal`), `-o` output file, `-` for stdin. Exit codes: `0` ok, `1` usage/IO error, `2` malformed JSON. ## Library use ```python import csv2json, json2csv header, rows = csv2json.read_rows(path, delimiter=None, encoding="utf-8-sig", has_header=True) objects = csv2json.rows_to_objects(header, rows, null_values=[""], as_number=False) records = json2csv.load_records(path) text = json2csv.to_csv(records, columns=json2csv.collect_columns(records), delimiter=",", quoting="minimal", missing="") ``` ## Tests `test_csv2json.py` (24 tests) and `test_json2csv.py` (20 tests) cover: * list-of-objects and keyed-object output shapes * delimiter auto-detection for `,` `;` tab `|` and explicit `--delimiter` * quoting/escaping of quotes, commas, tabs and embedded newlines * UTF-8 and BOM handling * null-token mapping, `--numbers`, `--missing` filler, bool rendering * nested values inlined as JSON instead of crashing * error paths: missing file, empty file, ragged row, non-array JSON, non-object row, invalid JSON, duplicate key column * stdin input and file output ```bash $ python3 -m pytest test_csv2json.py test_json2csv.py -q 44 passed ``` === csv2json.py === #!/usr/bin/env python3 """csv2json — convert CSV files of any structure to JSON. Zero dependencies (stdlib only), Python 3.8+. Features -------- * Auto-detects the delimiter (comma, semicolon, tab, pipe) from the header line. * Emits either a list of row objects (default) or a dict keyed by a chosen column. * ``--indent`` controls pretty-printing; ``--no-indent`` emits a compact single line. * ``--null-values`` maps given tokens (e.g. "", "NULL", "N/A") to JSON ``null``. * ``--numbers`` converts numeric-looking fields to JSON numbers. * ``--encoding`` for non-UTF-8 sources (default utf-8-sig, so BOM is stripped). * Clear, actionable errors for missing files, empty input and ragged rows. Exit codes: 0 ok, 1 usage/IO error, 2 malformed CSV. """ from __future__ import annotations import argparse import csv import json import sys from pathlib import Path from typing import Any, Dict, Iterable, List, Sequence SUPPORTED_DELIMITERS = [",", ";", "\t", "|"] DEFAULT_NULL_VALUES = ["", "NULL", "null", "N/A", "n/a", "-"] class MalformedCsvError(ValueError): """Raised when the CSV parses but its shape is inconsistent (e.g. ragged rows).""" def detect_delimiter(sample: str, candidates: Sequence[str] = SUPPORTED_DELIMITERS) -> str: """Pick the candidate delimiter that yields the most columns in the header.""" first_line = sample.splitlines()[0] if sample.strip() else "" best, best_score = ",", -1 for delim in candidates: score = len(next(csv.reader([first_line], delimiter=delim), [])) if score > best_score: best, best_score = delim, score return best def read_rows( path: Path, delimiter: str | None, encoding: str, has_header: bool, ) -> tuple[List[str], List[List[str]]]: """Return (header, rows). Raises ValueError with an actionable message.""" try: text = path.read_text(encoding=encoding) except FileNotFoundError: raise ValueError(f"file not found: {path}") except IsADirectoryError: raise ValueError(f"not a file: {path}") except UnicodeDecodeError as exc: raise ValueError( f"could not decode {path} as {encoding}; pass --encoding (e.g. latin-1)" ) from exc if not text.strip(): raise ValueError(f"file is empty: {path}") delim = delimiter or detect_delimiter(text) reader = csv.reader(text.splitlines(True), delimiter=delim) rows: List[List[str]] = [] for lineno, row in enumerate(reader, start=1): if not row or (len(row) == 1 and row[0].strip() == ""): continue # skip blank lines rows.append(row) if not rows: raise ValueError(f"no data rows in {path}") if has_header: header, body = rows[0], rows[1:] else: width = max(len(r) for r in rows) header = [f"col_{i + 1}" for i in range(width)] body = rows width = len(header) fixed: List[List[str]] = [] for lineno, row in enumerate(body, start=2 if has_header else 1): if len(row) != width: raise MalformedCsvError( f"{path}:{lineno}: expected {width} fields, found {len(row)} " f"(delimiter={delim!r}); check quoting or pass --delimiter" ) fixed.append(row) return header, fixed def convert_value(raw: str, null_values: Iterable[str], as_number: bool) -> Any: """Map one raw CSV cell to a JSON value.""" if raw in null_values: return None if as_number: try: return int(raw) except ValueError: try: return float(raw) except ValueError: return raw return raw def rows_to_objects( header: Sequence[str], rows: Sequence[Sequence[str]], null_values: Iterable[str], as_number: bool, ) -> List[Dict[str, Any]]: nulls = set(null_values) return [ {name: convert_value(value, nulls, as_number) for name, value in zip(header, row)} for row in rows ] def rows_to_keyed( header: Sequence[str], rows: Sequence[Sequence[str]], key_column: str, null_values: Iterable[str], as_number: bool, ) -> Dict[str, Any]: if key_column not in header: raise MalformedCsvError(f"--key-column {key_column!r} not in header {list(header)}") idx = header.index(key_column) out: Dict[str, Any] = {} for row in rows: key = row[idx] if key in out: raise MalformedCsvError(f"duplicate key {key!r} in column {key_column!r}; use list mode") out[key] = { name: convert_value(value, null_values, as_number) for name, value in zip(header, row) if name != key_column } return out def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="csv2json", description="Convert a CSV file to JSON (list of objects or keyed object).", ) p.add_argument("input", type=Path, help="input CSV file, or - for stdin") p.add_argument("-o", "--output", type=Path, default=None, help="output file (default: stdout)") p.add_argument("--delimiter", default=None, help="force delimiter (auto-detected by default)") p.add_argument("--encoding", default="utf-8-sig", help="input encoding (default: utf-8-sig)") p.add_argument("--no-header", action="store_true", help="treat the first row as data") p.add_argument("--key-column", default=None, help="emit an object keyed by this column") p.add_argument("--indent", type=int, default=2, help="indent spaces (default: 2)") p.add_argument("--compact", action="store_true", help="emit one compact line") p.add_argument("--null-values", nargs="*", default=DEFAULT_NULL_VALUES, help="tokens mapped to JSON null") p.add_argument("--numbers", action="store_true", help="convert numeric fields to JSON numbers") return p def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) if args.input == Path("-"): import tempfile with tempfile.NamedTemporaryFile("w", suffix=".csv", delete=False) as tmp: tmp.write(sys.stdin.read()) args.input = Path(tmp.name) try: header, rows = read_rows(args.input, args.delimiter, args.encoding, not args.no_header) except MalformedCsvError as exc: print(f"csv2json: error: {exc}", file=sys.stderr) return 2 except ValueError as exc: print(f"csv2json: error: {exc}", file=sys.stderr) return 1 try: if args.key_column: data = rows_to_keyed(header, rows, args.key_column, args.null_values, args.numbers) else: data = rows_to_objects(header, rows, args.null_values, args.numbers) except MalformedCsvError as exc: print(f"csv2json: error: {exc}", file=sys.stderr) return 2 if args.compact: text = json.dumps(data, ensure_ascii=False, separators=(",", ":")) else: text = json.dumps(data, ensure_ascii=False, indent=args.indent) if args.output: args.output.write_text(text + "\n", encoding="utf-8") print(f"csv2json: wrote {args.output}", file=sys.stderr) else: print(text) return 0 if __name__ == "__main__": raise SystemExit(main()) === json2csv.py === #!/usr/bin/env python3 """json2csv — convert a JSON array of flat objects to CSV. Zero dependencies (stdlib only), Python 3.10+. Features -------- * Reads a JSON array of flat objects from a file or stdin. * Collects the union of all keys; ``--columns`` pins an explicit order. * Missing keys become empty strings (``--missing`` overrides the filler). * ``--delimiter`` / ``--quoting`` control the emitted CSV dialect. * ``--sort-keys`` emits columns alphabetically for reproducible diffs. * Nested values (dict/list) are JSON-encoded inline rather than crashing. * Clear errors for non-array input, non-object rows and unreadable JSON. Exit codes: 0 ok, 1 usage/IO error, 2 malformed JSON. """ from __future__ import annotations import argparse import csv import io import json import sys from pathlib import Path from typing import Any, Dict, List, Sequence class MalformedJsonError(ValueError): """Raised when the JSON parses but its shape is unusable for CSV output.""" def load_records(path: Path) -> List[Any]: try: text = path.read_text(encoding="utf-8") except FileNotFoundError: raise ValueError(f"file not found: {path}") from None except IsADirectoryError: raise ValueError(f"not a file: {path}") from None except UnicodeDecodeError as exc: raise ValueError(f"could not decode {path} as utf-8") from exc if not text.strip(): raise MalformedJsonError(f"file is empty: {path}") try: data = json.loads(text) except json.JSONDecodeError as exc: raise MalformedJsonError(f"{path}: invalid JSON: {exc}") from exc if not isinstance(data, list): raise MalformedJsonError( f"{path}: expected a JSON array of objects, got {type(data).__name__}" ) return data def collect_columns(records: Sequence[Any]) -> List[str]: columns: List[str] = [] seen: set[str] = set() for i, rec in enumerate(records, start=1): if not isinstance(rec, dict): raise MalformedJsonError(f"record {i} is a {type(rec).__name__}, expected an object") for key in rec: if key not in seen: seen.add(key) columns.append(key) return columns def render(value: Any) -> str: """Render one JSON value as a CSV cell.""" if value is None: return "" if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (int, float, str)): return str(value) return json.dumps(value, ensure_ascii=False, separators=(",", ":")) def to_csv( records: Sequence[Any], columns: Sequence[str], delimiter: str, quoting: str, missing: str, ) -> str: if not columns: return "" buf = io.StringIO() if quoting == "all": writer = csv.writer(buf, delimiter=delimiter, quoting=csv.QUOTE_ALL, lineterminator="\n") elif quoting == "minimal": writer = csv.writer(buf, delimiter=delimiter, quoting=csv.QUOTE_MINIMAL, lineterminator="\n") else: # nonnumeric writer = csv.writer(buf, delimiter=delimiter, quoting=csv.QUOTE_NONNUMERIC, lineterminator="\n") writer.writerow(columns) for rec in records: writer.writerow([ missing if col not in rec else render(rec[col]) for col in columns ]) return buf.getvalue() def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="json2csv", description="Convert a JSON array of flat objects to CSV.", ) p.add_argument("input", type=Path, help="input JSON file, or - for stdin") p.add_argument("-o", "--output", type=Path, default=None, help="output file (default: stdout)") p.add_argument("--columns", nargs="+", default=None, help="explicit column order") p.add_argument("--delimiter", default=",", help="output delimiter (default: ,)") p.add_argument("--quoting", choices=["minimal", "all", "nonnumeric"], default="minimal") p.add_argument("--missing", default="", help="filler for absent keys (default: empty string)") p.add_argument("--sort-keys", action="store_true", help="emit columns alphabetically") return p def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) if args.input == Path("-"): import tempfile with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as tmp: tmp.write(sys.stdin.read()) args.input = Path(tmp.name) try: records = load_records(args.input) except MalformedJsonError as exc: print(f"json2csv: error: {exc}", file=sys.stderr) return 2 except ValueError as exc: print(f"json2csv: error: {exc}", file=sys.stderr) return 1 try: columns = args.columns or collect_columns(records) if args.sort_keys: columns = sorted(columns) except MalformedJsonError as exc: print(f"json2csv: error: {exc}", file=sys.stderr) return 2 text = to_csv(records, columns, args.delimiter, args.quoting, args.missing) if args.output: args.output.write_text(text, encoding="utf-8") print(f"json2csv: wrote {args.output}", file=sys.stderr) else: sys.stdout.write(text) return 0 if __name__ == "__main__": raise SystemExit(main()) === test_csv2json.py === """Tests for csv2json.""" import json import subprocess import sys from pathlib import Path import pytest SCRIPT = Path(__file__).with_name("csv2json.py") def run(*args, stdin=None): return subprocess.run( [sys.executable, str(SCRIPT), *args], input=stdin, capture_output=True, text=True, ) def write(tmp_path, name, content): p = tmp_path / name p.write_text(content, encoding="utf-8") return p # ---------------------------------------------------------------- basic def test_basic_list_of_objects(tmp_path): src = write(tmp_path, "a.csv", "name,city\nAndy,Jakarta\nBudi,Bandung\n") r = run(str(src)) assert r.returncode == 0, r.stderr assert json.loads(r.stdout) == [ {"name": "Andy", "city": "Jakarta"}, {"name": "Budi", "city": "Bandung"}, ] def test_special_characters_are_escaped(tmp_path): src = write(tmp_path, "a.csv", 'note\n"has ""quotes"", comma, and\ttab"\nplain\n') r = run(str(src)) assert r.returncode == 0, r.stderr data = json.loads(r.stdout) assert data[0]["note"] == 'has "quotes", comma, and\ttab' assert data[1]["note"] == "plain" def test_unicode_roundtrip(tmp_path): src = write(tmp_path, "a.csv", "k\n日本\ncafé\n") r = run(str(src)) assert json.loads(r.stdout) == [{"k": "日本"}, {"k": "café"}] def test_bom_is_stripped(tmp_path): p = tmp_path / "bom.csv" p.write_bytes("\ufeffname\nAndy\n".encode("utf-8")) r = run(str(p)) assert json.loads(r.stdout) == [{"name": "Andy"}] # ------------------------------------------------------- delimiter detection @pytest.mark.parametrize("delim", [";", "\t", "|"]) def test_delimiter_autodetect(tmp_path, delim): src = write(tmp_path, "d.csv", f"a{delim}b\n1{delim}2\n") r = run(str(src)) assert json.loads(r.stdout) == [{"a": "1", "b": "2"}] def test_explicit_delimiter_overrides(tmp_path): src = write(tmp_path, "d.csv", "a;b\n1;2\n") r = run(str(src), "--delimiter", ";") assert json.loads(r.stdout) == [{"a": "1", "b": "2"}] # ------------------------------------------------------------- value mapping def test_null_values(tmp_path): src = write(tmp_path, "a.csv", "x,y\n1,\n2,NULL\n3,N/A\n") r = run(str(src)) assert json.loads(r.stdout) == [ {"x": "1", "y": None}, {"x": "2", "y": None}, {"x": "3", "y": None}, ] def test_numbers_flag(tmp_path): src = write(tmp_path, "a.csv", "i,f,s\n42,3.14,abc\n") r = run(str(src), "--numbers") assert json.loads(r.stdout) == [{"i": 42, "f": 3.14, "s": "abc"}] def test_custom_null_values(tmp_path): src = write(tmp_path, "a.csv", "x\nMISSING\n1\n") r = run(str(src), "--null-values", "MISSING") assert json.loads(r.stdout) == [{"x": None}, {"x": "1"}] # --------------------------------------------------------------- output shape def test_keyed_object(tmp_path): src = write(tmp_path, "a.csv", "id,v\n1,a\n2,b\n") r = run(str(src), "--key-column", "id") assert json.loads(r.stdout) == {"1": {"v": "a"}, "2": {"v": "b"}} def test_keyed_duplicate_is_error(tmp_path): src = write(tmp_path, "a.csv", "id,v\n1,a\n1,b\n") r = run(str(src), "--key-column", "id") assert r.returncode == 2 assert "duplicate key" in r.stderr def test_key_column_missing_is_error(tmp_path): src = write(tmp_path, "a.csv", "id,v\n1,a\n") r = run(str(src), "--key-column", "nope") assert r.returncode == 2 def test_no_header(tmp_path): src = write(tmp_path, "a.csv", "Andy,Jakarta\n") r = run(str(src), "--no-header") assert json.loads(r.stdout) == [{"col_1": "Andy", "col_2": "Jakarta"}] def test_compact_output(tmp_path): src = write(tmp_path, "a.csv", "a\n1\n") r = run(str(src), "--compact") assert r.stdout.strip() == '[{"a":"1"}]' def test_indent_option(tmp_path): src = write(tmp_path, "a.csv", "a\n1\n") r = run(str(src), "--indent", "4") assert '\n "a"' in r.stdout def test_write_to_file(tmp_path): src = write(tmp_path, "a.csv", "a\n1\n") out = tmp_path / "out.json" r = run(str(src), "-o", str(out)) assert r.returncode == 0 assert json.loads(out.read_text()) == [{"a": "1"}] # ----------------------------------------------------------------- errors def test_missing_file(tmp_path): r = run(str(tmp_path / "nope.csv")) assert r.returncode == 1 assert "file not found" in r.stderr def test_empty_file(tmp_path): src = write(tmp_path, "a.csv", "") r = run(str(src)) assert r.returncode == 1 assert "empty" in r.stderr def test_ragged_row_is_error(tmp_path): src = write(tmp_path, "a.csv", "a,b\n1,2\n3\n") r = run(str(src)) assert r.returncode == 2 assert "expected 2 fields, found 1" in r.stderr def test_stdin_input(tmp_path): r = run("-", stdin="a\n1\n") assert r.returncode == 0 assert json.loads(r.stdout) == [{"a": "1"}] def test_blank_lines_skipped(tmp_path): src = write(tmp_path, "a.csv", "a\n1\n\n2\n") r = run(str(src)) assert json.loads(r.stdout) == [{"a": "1"}, {"a": "2"}] # ------------------------------------------------------------------ library API def test_library_import(tmp_path): sys.path.insert(0, str(SCRIPT.parent)) import csv2json # noqa: PLC0415 header, rows = csv2json.read_rows(write(tmp_path, "a.csv", "a,b\n1,2\n"), None, "utf-8-sig", True) assert csv2json.rows_to_objects(header, rows, [""], False) == [{"a": "1", "b": "2"}] === test_json2csv.py === """Tests for json2csv.""" import json import subprocess import sys from pathlib import Path SCRIPT = Path(__file__).with_name("json2csv.py") def run(*args, stdin=None): return subprocess.run( [sys.executable, str(SCRIPT), *args], input=stdin, capture_output=True, text=True, ) def write(tmp_path, name, obj): p = tmp_path / name p.write_text(json.dumps(obj), encoding="utf-8") return p def parse(text, delimiter=","): import csv import io return list(csv.reader(io.StringIO(text), delimiter=delimiter)) def test_basic_conversion(tmp_path): src = write(tmp_path, "a.json", [{"a": 1, "b": "x"}, {"a": 2, "b": "y"}]) r = run(str(src)) assert r.returncode == 0, r.stderr assert parse(r.stdout) == [["a", "b"], ["1", "x"], ["2", "y"]] def test_missing_keys_filled_with_empty_string(tmp_path): src = write(tmp_path, "a.json", [{"a": 1}, {"a": 2, "b": 3}]) r = run(str(src)) assert parse(r.stdout) == [["a", "b"], ["1", ""], ["2", "3"]] def test_missing_filler_override(tmp_path): src = write(tmp_path, "a.json", [{"a": 1}, {"a": 2, "b": 3}]) r = run(str(src), "--missing", "N/A") assert parse(r.stdout) == [["a", "b"], ["1", "N/A"], ["2", "3"]] def test_null_becomes_empty(tmp_path): src = write(tmp_path, "a.json", [{"a": None, "b": 1}]) r = run(str(src)) assert parse(r.stdout) == [["a", "b"], ["", "1"]] def test_bool_rendering(tmp_path): src = write(tmp_path, "a.json", [{"ok": True, "no": False}]) r = run(str(src)) assert parse(r.stdout) == [["ok", "no"], ["true", "false"]] def test_special_characters_are_quoted(tmp_path): src = write(tmp_path, "a.json", [{"note": 'has "quotes", comma'}]) r = run(str(src)) rows = parse(r.stdout) assert rows[1][0] == 'has "quotes", comma' def test_custom_delimiter(tmp_path): src = write(tmp_path, "a.json", [{"a": 1, "b": 2}]) r = run(str(src), "--delimiter", ";") assert parse(r.stdout, ";") == [["a", "b"], ["1", "2"]] def test_explicit_columns(tmp_path): src = write(tmp_path, "a.json", [{"a": 1, "b": 2, "c": 3}]) r = run(str(src), "--columns", "c", "a") assert parse(r.stdout) == [["c", "a"], ["3", "1"]] def test_sort_keys(tmp_path): src = write(tmp_path, "a.json", [{"z": 1, "a": 2, "m": 3}]) r = run(str(src), "--sort-keys") assert parse(r.stdout) == [["a", "m", "z"], ["2", "3", "1"]] def test_nested_value_is_inlined_json(tmp_path): src = write(tmp_path, "a.json", [{"tags": ["x", "y"], "n": 1}]) r = run(str(src)) assert parse(r.stdout)[1][0] == '["x","y"]' def test_quoting_all(tmp_path): src = write(tmp_path, "a.json", [{"a": 1}]) r = run(str(src), "--quoting", "all") assert r.stdout == '"a"\n"1"\n' # and the quotes are real CSV quoting, not literal characters assert parse(r.stdout) == [["a"], ["1"]] def test_empty_array_header_only(tmp_path): src = write(tmp_path, "a.json", []) r = run(str(src)) assert r.returncode == 0 assert r.stdout == "" def test_write_to_file(tmp_path): src = write(tmp_path, "a.json", [{"a": 1}]) out = tmp_path / "out.csv" r = run(str(src), "-o", str(out)) assert r.returncode == 0 assert parse(out.read_text()) == [["a"], ["1"]] def test_stdin_input(tmp_path): r = run("-", stdin='[{"a":1}]') assert r.returncode == 0 assert parse(r.stdout) == [["a"], ["1"]] def test_missing_file(tmp_path): r = run(str(tmp_path / "nope.json")) assert r.returncode == 1 assert "file not found" in r.stderr def test_not_an_array(tmp_path): src = tmp_path / "obj.json" src.write_text('{"a":1}') r = run(str(src)) assert r.returncode == 2 assert "expected a JSON array" in r.stderr def test_non_object_row(tmp_path): src = tmp_path / "bad.json" src.write_text('[{"a":1}, "oops"]') r = run(str(src)) assert r.returncode == 2 assert "expected an object" in r.stderr def test_invalid_json(tmp_path): src = tmp_path / "bad.json" src.write_text("{not json") r = run(str(src)) assert r.returncode == 2 assert "invalid JSON" in r.stderr def test_empty_file(tmp_path): src = tmp_path / "e.json" src.write_text("") r = run(str(src)) assert r.returncode == 2 assert "empty" in r.stderr def test_library_import(tmp_path): sys.path.insert(0, str(SCRIPT.parent)) import json2csv # noqa: PLC0415 assert json2csv.collect_columns([{"a": 1}, {"b": 2}]) == ["a", "b"] assert json2csv.render(None) == "" assert json2csv.render({"k": 1}) == '{"k":1}' === VERIFIED RUN (live output) === $ printf "name,city,note" > demo.csv ... $ python3 csv2json.py demo.csv [ { "name": "Andi, Jr", "city": "Jakarta", "note": "says \"hi\"" }, { "name": "Budi", "city": "Bandung", "note": "plain" } ] $ printf "[{\"a\":1,\"b\":\"x\"},{\"a\":2}]" | python3 json2csv.py /dev/stdin a,b 1,x 2, $ printf "a,b\n1\n" | python3 csv2json.py /dev/stdin # ragged row exit=2 $ python3 -m pytest -q ............................................ [100%] 44 passed in 6.66s