Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/services/memory.py: 100%

1# SPDX-License-Identifier: MIT 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Provides a thread-safe, file-persisted key-value store. 

5 

6This module defines the `Memory` class, a concrete implementation of the 

7`MemoryProtocol`. It uses a dictionary for in-memory storage, protected by a 

8`threading.Lock` for thread safety. Unlike a purely transient store, this 

9implementation persists the entire key-value store to a JSON file on every 

10write operation, allowing state to survive across different CLI invocations. 

11""" 

12 

13from __future__ import annotations 

14 

15import json 

16from threading import Lock 

17from typing import Any 

18 

19from injector import inject 

20 

21from bijux_cli.contracts import MemoryProtocol 

22from bijux_cli.core.paths import MEMORY_FILE 

23 

24 

25class Memory(MemoryProtocol): 

26 """Implements `MemoryProtocol` with a thread-safe, file-backed dictionary. 

27 

28 This service provides a simple key-value store that is both thread-safe 

29 and persistent to a JSON file (`~/.bijux/.memory.json`). 

30 

31 Attributes: 

32 _store (dict[str, Any]): The in-memory dictionary holding the data. 

33 _lock (Lock): A lock to ensure thread-safe access to the store. 

34 """ 

35 

36 @inject 

37 def __init__(self) -> None: 

38 """Initializes the service, loading existing data from the persistence file.""" 

39 MEMORY_FILE.parent.mkdir(parents=True, exist_ok=True) 

40 try: 

41 with MEMORY_FILE.open("r") as f: 

42 self._store: dict[str, Any] = json.load(f) 

43 except (FileNotFoundError, json.JSONDecodeError): 

44 self._store = {} 

45 self._lock = Lock() 

46 

47 def get(self, key: str) -> Any: 

48 """Retrieves a value by its key in a thread-safe manner. 

49 

50 Args: 

51 key (str): The key of the value to retrieve. 

52 

53 Returns: 

54 Any: The value associated with the key. 

55 

56 Raises: 

57 KeyError: If the key does not exist in the store. 

58 """ 

59 with self._lock: 

60 if key not in self._store: 

61 raise KeyError(f"Memory key not found: {key}") 

62 return self._store[key] 

63 

64 def set(self, key: str, value: Any) -> None: 

65 """Sets a key-value pair and persists the change to disk. 

66 

67 If the key already exists, its value is overwritten. This operation 

68 is thread-safe. 

69 

70 Args: 

71 key (str): The key for the value being set. 

72 value (Any): The value to store. 

73 

74 Returns: 

75 None: 

76 """ 

77 with self._lock: 

78 self._store[key] = value 

79 self._persist() 

80 

81 def delete(self, key: str) -> None: 

82 """Deletes a key-value pair and persists the change to disk. 

83 

84 This operation is thread-safe. 

85 

86 Args: 

87 key (str): The key of the value to delete. 

88 

89 Raises: 

90 KeyError: If the key does not exist in the store. 

91 """ 

92 with self._lock: 

93 if key not in self._store: 

94 raise KeyError(f"Memory key not found: {key}") 

95 del self._store[key] 

96 self._persist() 

97 

98 def clear(self) -> None: 

99 """Removes all key-value pairs and persists the change to disk. 

100 

101 This operation is thread-safe. 

102 """ 

103 with self._lock: 

104 self._store.clear() 

105 self._persist() 

106 

107 def keys(self) -> list[str]: 

108 """Returns a list of all keys currently in the store. 

109 

110 This operation is thread-safe. 

111 

112 Returns: 

113 list[str]: A list of all string keys. 

114 """ 

115 with self._lock: 

116 return list(self._store.keys()) 

117 

118 def _persist(self) -> None: 

119 """Writes the current in-memory store to the JSON persistence file. 

120 

121 Note: 

122 This method is not thread-safe on its own and should only be 

123 called from within a block that holds `self._lock`. 

124 """ 

125 with MEMORY_FILE.open("w") as f: 

126 json.dump(self._store, f) 

127 

128 

129__all__ = ["Memory"]