Coverage for src \ truenex_memory \ export \ __init__.py: 91%

1"""JSON export and import helpers for local memory payloads.""" 

2 

3from __future__ import annotations 

4 

5import json 

6from datetime import datetime, timezone 

7from pathlib import Path 

8from typing import Any, Iterable, Mapping 

9 

10SCHEMA_VERSION = 1 

11 

12 

13def build_export_payload( 

14 records: Iterable[Mapping[str, Any]], 

15 *, 

16 metadata: Mapping[str, Any] | None = None, 

17) -> dict[str, Any]: 

18 """Build a stable JSON payload for memory records.""" 

19 

20 return { 

21 "schema_version": SCHEMA_VERSION, 

22 "exported_at": datetime.now(timezone.utc).isoformat(), 

23 "metadata": dict(metadata or {}), 

24 "records": [dict(record) for record in records], 

25 } 

26 

27 

28def export_json( 

29 records: Iterable[Mapping[str, Any]], 

30 destination: str | Path, 

31 *, 

32 metadata: Mapping[str, Any] | None = None, 

33) -> dict[str, Any]: 

34 """Write records to a local JSON export file and return the payload.""" 

35 

36 payload = build_export_payload(records, metadata=metadata) 

37 path = Path(destination) 

38 path.parent.mkdir(parents=True, exist_ok=True) 

39 path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") 

40 return payload 

41 

42 

43def import_json(source: str | Path) -> dict[str, Any]: 

44 """Read and validate a local JSON export file.""" 

45 

46 path = Path(source) 

47 payload = json.loads(path.read_text(encoding="utf-8")) 

48 if not isinstance(payload, dict): 

49 raise ValueError("export payload must be a JSON object") 

50 if payload.get("schema_version") != SCHEMA_VERSION: 

51 raise ValueError(f"unsupported schema_version: {payload.get('schema_version')!r}") 

52 records = payload.get("records") 

53 if not isinstance(records, list): 

54 raise ValueError("export payload must contain a records list") 

55 for index, record in enumerate(records): 

56 if not isinstance(record, dict): 

57 raise ValueError(f"record {index} must be a JSON object") 

58 payload.setdefault("metadata", {}) 

59 return payload 

60 

61 

62def import_records(source: str | Path) -> list[dict[str, Any]]: 

63 """Read records from a local JSON export file.""" 

64 

65 return list(import_json(source)["records"]) 

66 

67 

68from truenex_memory.export.fingerprint import canonicalize_export, export_fingerprint, exports_equivalent 

69 

70 

71__all__ = [ 

72 "SCHEMA_VERSION", 

73 "build_export_payload", 

74 "canonicalize_export", 

75 "export_fingerprint", 

76 "export_json", 

77 "exports_equivalent", 

78 "import_json", 

79 "import_records", 

80]