diff --git a/hef/catalog.py b/hef/catalog.py index d4bcc10..56aac04 100644 --- a/hef/catalog.py +++ b/hef/catalog.py @@ -5,6 +5,9 @@ from __future__ import annotations from dataclasses import dataclass, asdict, fields from typing import Optional +import json +from pathlib import Path + MODES = frozenset({"audio", "video", "av"}) LICENSES = frozenset({"public_domain", "cc0", "cc_by", "cc_by_nc"}) REVIEW_STATUSES = frozenset({"proposed", "approved"}) @@ -87,3 +90,39 @@ def record_from_dict(data: dict) -> Record: return Record(**data) except TypeError as exc: raise CatalogError(str(exc)) from exc + + +def load_catalog(path) -> list[Record]: + """Load and validate every record from a JSONL file. Blank lines skipped.""" + path = Path(path) + records: list[Record] = [] + with path.open("r", encoding="utf-8") as fh: + for lineno, raw in enumerate(fh, start=1): + line = raw.strip() + if not line: + continue + try: + data = json.loads(line) + except json.JSONDecodeError as exc: + raise CatalogError(f"line {lineno}: invalid JSON: {exc}") from exc + record = record_from_dict(data) + validate(record) + records.append(record) + return records + + +def save_catalog(records, path) -> None: + """Validate and write all records to a JSONL file (overwrites).""" + path = Path(path) + with path.open("w", encoding="utf-8") as fh: + for record in records: + validate(record) + fh.write(json.dumps(record_to_dict(record), ensure_ascii=False) + "\n") + + +def append_record(record, path) -> None: + """Validate and append a single record to a JSONL file.""" + validate(record) + path = Path(path) + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(record_to_dict(record), ensure_ascii=False) + "\n") diff --git a/tests/test_catalog.py b/tests/test_catalog.py index 6ed5f13..b23b721 100644 --- a/tests/test_catalog.py +++ b/tests/test_catalog.py @@ -92,3 +92,62 @@ def test_record_from_dict_rejects_missing_required_field(): del data["title"] with pytest.raises(CatalogError): record_from_dict(data) + + +from hef.catalog import load_catalog, save_catalog, append_record + + +def test_save_then_load_round_trip(tmp_path): + path = tmp_path / "library.jsonl" + records = [make_record(id="a"), make_record(id="b", mode="audio")] + save_catalog(records, path) + loaded = load_catalog(path) + assert loaded == records + + +def test_load_skips_blank_lines(tmp_path): + path = tmp_path / "library.jsonl" + save_catalog([make_record(id="a")], path) + with path.open("a", encoding="utf-8") as fh: + fh.write("\n \n") + loaded = load_catalog(path) + assert len(loaded) == 1 + + +def test_load_empty_file_returns_empty_list(tmp_path): + path = tmp_path / "library.jsonl" + path.write_text("", encoding="utf-8") + assert load_catalog(path) == [] + + +def test_load_invalid_json_raises(tmp_path): + path = tmp_path / "library.jsonl" + path.write_text("{not json}\n", encoding="utf-8") + with pytest.raises(CatalogError): + load_catalog(path) + + +def test_load_validates_records(tmp_path): + path = tmp_path / "library.jsonl" + save_catalog([make_record(id="a")], path) + import json + bad = record_to_dict(make_record(id="bad")) + bad["left"] = 9 + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(bad) + "\n") + with pytest.raises(CatalogError): + load_catalog(path) + + +def test_append_record(tmp_path): + path = tmp_path / "library.jsonl" + save_catalog([make_record(id="a")], path) + append_record(make_record(id="b"), path) + loaded = load_catalog(path) + assert [r.id for r in loaded] == ["a", "b"] + + +def test_save_rejects_invalid_record(tmp_path): + path = tmp_path / "library.jsonl" + with pytest.raises(CatalogError): + save_catalog([make_record(mode="hologram")], path)