Coverage for merco/memory/backends/json_backend.py: 100%

1"""JSONBackend — 每记忆一个 .json 文件""" 

2from __future__ import annotations 

3 

4import json 

5from datetime import datetime 

6from pathlib import Path 

7 

8from merco.memory.backend import MemoryBackend 

9 

10 

11class JSONBackend(MemoryBackend): 

12 """JSON 文件后端""" 

13 name = "json" 

14 

15 def __init__(self, base_path: str): 

16 self.base_path = Path(base_path).expanduser() 

17 self.base_path.mkdir(parents=True, exist_ok=True) 

18 

19 def save(self, key: str, value, tags: list = None) -> None: 

20 entry = { 

21 "key": key, 

22 "value": value, 

23 "tags": tags or [], 

24 "created_at": datetime.now().isoformat(), 

25 } 

26 file_path = self.base_path / f"{key}.json" 

27 with open(file_path, "w") as f: 

28 json.dump(entry, f, indent=2) 

29 

30 def load(self, key: str) -> dict | None: 

31 file_path = self.base_path / f"{key}.json" 

32 if not file_path.exists(): 

33 return None 

34 with open(file_path) as f: 

35 return json.load(f) 

36 

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

38 file_path = self.base_path / f"{key}.json" 

39 if file_path.exists(): 

40 file_path.unlink() 

41 

42 def list_keys(self, tag: str = None) -> list[str]: 

43 keys = [] 

44 for f in self.base_path.glob("*.json"): 

45 if tag: 

46 with open(f) as fh: 

47 data = json.load(fh) 

48 if tag in data.get("tags", []): 

49 keys.append(f.stem) 

50 else: 

51 keys.append(f.stem) 

52 return keys 

53 

54 def search(self, query: str) -> list[dict]: 

55 results = [] 

56 query_lower = query.lower() 

57 for f in self.base_path.glob("*.json"): 

58 with open(f) as fh: 

59 data = json.load(fh) 

60 content = json.dumps(data).lower() 

61 if query_lower in content: 

62 results.append(data) 

63 return results