Coverage for src/lilbee/app/reset.py: 100%

1"""Knowledge base reset (delete all documents and data).""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import shutil 

7from pathlib import Path 

8 

9from pydantic import BaseModel 

10 

11from lilbee.core import settings 

12from lilbee.core.config import cfg 

13from lilbee.data.store.held_import import remove_legacy_files 

14from lilbee.runtime.lock import library_held_for_reset 

15 

16 

17class ResetResult(BaseModel): 

18 """Result of a full knowledge base reset.""" 

19 

20 command: str = "reset" 

21 deleted_docs: int 

22 deleted_data: int 

23 skipped: list[str] = [] 

24 documents_dir: str 

25 data_dir: str 

26 

27 

28def _clear_dir(base_dir: Path, skipped: list[str]) -> int: 

29 """Delete all items in *base_dir*, appending undeletable paths to *skipped*.""" 

30 log = logging.getLogger(__name__) 

31 deleted = 0 

32 if not base_dir.exists(): 

33 return deleted 

34 for item in list(base_dir.iterdir()): 

35 try: 

36 # iterdir yields direct children only, so the entry is within 

37 # base_dir by construction. Registered source roots live outside this 

38 # dir (only their config entry is here), so reset never reaches a 

39 # user's corpus: it deletes owned files, and perform_reset 

40 # un-registers the roots from config.toml separately. 

41 if item.is_dir(): 

42 shutil.rmtree(item) 

43 else: 

44 item.unlink() 

45 except OSError as exc: 

46 # best-effort: reset is "delete as much as you can", and the 

47 # caller surfaces the skipped list to the user verbatim. 

48 log.warning("Could not delete %s: %s", item, exc) 

49 skipped.append(str(item)) 

50 continue 

51 deleted += 1 

52 return deleted 

53 

54 

55def perform_reset() -> ResetResult: 

56 """Delete all documents and data and un-register every linked source. 

57 

58 Raises ``ResetRefusedError`` while a sync, an import, a wiki build or an add that 

59 moves source keys, in this process or another, runs against the same data root, 

60 or when the lock that shows one cannot be taken: any of them would write back 

61 what the reset removed. 

62 """ 

63 skipped: list[str] = [] 

64 with library_held_for_reset(cfg): 

65 deleted_docs = _clear_dir(cfg.documents_dir, skipped) 

66 deleted_data = _clear_dir(cfg.data_dir, skipped) 

67 

68 # config.toml lives at data_root, next to (not inside) the two cleared 

69 # dirs. Without un-registering the roots here, in the file AND in this 

70 # process, the next sync walks every linked root and re-indexes it, so 

71 # content reappears after a "factory reset". Other settings survive: 

72 # reset deletes data, not configuration. 

73 settings.delete_value(cfg.data_root, "linked_roots") 

74 settings.sync_from_resolver(cfg, ["linked_roots"]) 

75 # The held-out record files of an older lilbee sit at data_root too; left 

76 # there, the next open would import them into the emptied index. 

77 remove_legacy_files(cfg.data_root) 

78 

79 return ResetResult( 

80 deleted_docs=deleted_docs, 

81 deleted_data=deleted_data, 

82 skipped=skipped, 

83 documents_dir=str(cfg.documents_dir), 

84 data_dir=str(cfg.data_dir), 

85 )