Coverage for src / lexigram / admin / state / persistence.py: 0%

1"""State persistence for Lexigram Admin.""" 

2 

3from __future__ import annotations 

4 

5import contextlib 

6from typing import Any, Protocol 

7 

8from lexigram import serialization as json 

9from lexigram.admin.state.store import Signal, watch 

10 

11 

12class IStorage(Protocol): 

13 """Protocol for storage backends.""" 

14 

15 def get(self, key: str) -> str | None: ... 

16 

17 def set(self, key: str, value: str) -> None: ... 

18 

19 def remove(self, key: str) -> None: ... 

20 

21 

22class DictionaryStorage(IStorage): 

23 """Memory-only storage for testing or server-side usage.""" 

24 

25 def __init__(self, initial_data: dict[str, str] | None = None) -> None: 

26 self._data = dict(initial_data or {}) 

27 

28 def get(self, key: str) -> str | None: 

29 return self._data.get(key) 

30 

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

32 self._data[key] = value 

33 

34 def remove(self, key: str) -> None: 

35 self._data.pop(key, None) 

36 

37 

38class PersistedSignal(Signal[Any]): 

39 """A signal that automatically persists its value to a storage backend.""" 

40 

41 def __init__(self, key: str, initial_value: Any, storage: IStorage) -> None: 

42 self._key = key 

43 self._storage = storage 

44 

45 # Try to load from storage 

46 stored_val = storage.get(key) 

47 if stored_val is not None: 

48 from lexigram.serialization import loads_str 

49 

50 try: 

51 value = loads_str(stored_val) 

52 except (ValueError, TypeError, json.JSONDecodeError): 

53 value = initial_value 

54 else: 

55 value = initial_value 

56 

57 super().__init__(value) 

58 

59 # Set up auto-persistence 

60 @watch 

61 def _persist() -> Any: 

62 val = self.get() # Register dependency 

63 from lexigram.serialization import dumps_str 

64 

65 self._storage.set(self._key, dumps_str(val)) 

66 

67 

68class StatePersistenceManager: 

69 """Manages persistence of multiple signals.""" 

70 

71 def __init__(self, storage: IStorage) -> None: 

72 self.storage = storage 

73 

74 def persist(self, key: str, signal: Signal[Any]) -> None: 

75 """Add persistence to an existing signal.""" 

76 # Load initial value 

77 stored_val = self.storage.get(key) 

78 if stored_val is not None: 

79 from lexigram.serialization import loads_str 

80 

81 with contextlib.suppress(Exception): 

82 signal.set(loads_str(stored_val)) 

83 

84 # Watch for changes and save 

85 @watch 

86 def _save() -> Any: 

87 val = signal.get() 

88 from lexigram.serialization import dumps_str 

89 

90 self.storage.set(key, dumps_str(val))