Added write and read history file functions

pkgs/clan-cli/clan_cli/history/add.py

@@ -9,7 +9,7 @@ from clan_cli.flakes.inspect import FlakeConfig, inspect_flake
 
 from ..clan_uri import ClanURI
 from ..dirs import user_history_file
-from ..locked_open import locked_open
+from ..locked_open import read_history_file, write_history_file
 
 
 class EnhancedJSONEncoder(json.JSONEncoder):
@@ -34,14 +34,12 @@ def list_history() -> list[HistoryEntry]:
     if not user_history_file().exists():
         return []
 
-    with locked_open(user_history_file(), "r") as f:
-        try:
-            content: str = f.read()
-            parsed: list[dict] = json.loads(content)
-            logs = [HistoryEntry(**p) for p in parsed]
-        except (json.JSONDecodeError, TypeError) as ex:
-            print("Failed to load history. Invalid JSON.")
-            print(f"{user_history_file()}: {ex}")
+    try:
+        parsed = read_history_file()
+        logs = [HistoryEntry(**p) for p in parsed]
+    except (json.JSONDecodeError, TypeError) as ex:
+        print("Failed to load history. Invalid JSON.")
+        print(f"{user_history_file()}: {ex}")
 
     return logs
 
@@ -69,9 +67,7 @@ def add_history(uri: ClanURI) -> list[HistoryEntry]:
         )
         logs.append(history)
 
-    with locked_open(user_history_file(), "w+") as f:
-        f.write(json.dumps(logs, cls=EnhancedJSONEncoder, indent=4))
-        f.truncate()
+    write_history_file(logs)
 
     return logs
 

pkgs/clan-cli/clan_cli/history/update.py

@@ -2,12 +2,10 @@
 import argparse
 import copy
 import datetime
-import json
 
-from ..dirs import user_history_file
-from ..locked_open import locked_open
+from ..locked_open import write_history_file
 from ..nix import nix_metadata
-from .add import EnhancedJSONEncoder, HistoryEntry, list_history
+from .add import HistoryEntry, list_history
 
 
 def update_history() -> list[HistoryEntry]:
@@ -29,9 +27,7 @@ def update_history() -> list[HistoryEntry]:
         # TODO: Delete stale entries
         new_logs.append(new_entry)
 
-    with locked_open(user_history_file(), "w+") as f:
-        f.write(json.dumps(new_logs, cls=EnhancedJSONEncoder, indent=4))
-        f.truncate()
+    write_history_file(new_logs)
     return new_logs
 
 

pkgs/clan-cli/clan_cli/locked_open.py

@@ -1,11 +1,23 @@
+import dataclasses
 import fcntl
+import json
 from collections.abc import Generator
 from contextlib import contextmanager
 from pathlib import Path
+from typing import Any
+
+from .dirs import user_history_file
+
+
+class EnhancedJSONEncoder(json.JSONEncoder):
+    def default(self, o: Any) -> Any:
+        if dataclasses.is_dataclass(o):
+            return dataclasses.asdict(o)
+        return super().default(o)
 
 
 @contextmanager
-def locked_open(filename: str | Path, mode: str = "r") -> Generator:
+def _locked_open(filename: str | Path, mode: str = "r") -> Generator:
     """
     This is a context manager that provides an advisory write lock on the file specified by `filename` when entering the context, and releases the lock when leaving the context. The lock is acquired using the `fcntl` module's `LOCK_EX` flag, which applies an exclusive write lock to the file.
     """
@@ -13,3 +25,16 @@ def locked_open(filename: str | Path, mode: str = "r") -> Generator:
         fcntl.flock(fd, fcntl.LOCK_EX)
         yield fd
         fcntl.flock(fd, fcntl.LOCK_UN)
+
+
+def write_history_file(data: Any) -> None:
+    with _locked_open(user_history_file(), "w+") as f:
+        f.write(json.dumps(data, cls=EnhancedJSONEncoder, indent=4))
+        f.truncate()
+
+
+def read_history_file() -> list[dict]:
+    with _locked_open(user_history_file(), "r") as f:
+        content: str = f.read()
+        parsed: list[dict] = json.loads(content)
+        return parsed