- # 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