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.ingest.skip_marker import clear_skip_markers 

14from lilbee.runtime.lock import no_sync_running 

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 or import, in this process or another, 

59 runs against the same data root, or when the lock that shows one cannot be taken: 

60 a running sync or import would write back what the reset removed. 

61 """ 

62 skipped: list[str] = [] 

63 with no_sync_running(cfg.data_root): 

64 deleted_docs = _clear_dir(cfg.documents_dir, skipped) 

65 deleted_data = _clear_dir(cfg.data_dir, skipped) 

66 

67 # config.toml and the skip sidecars live at data_root, next to (not inside) 

68 # the two cleared dirs. Without un-registering the roots here, in the file 

69 # AND in this process, the next sync walks every linked root and re-indexes 

70 # it, so content reappears after a "factory reset". Other settings survive: 

71 # reset deletes data, not configuration. 

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

73 cfg.linked_roots = {} 

74 clear_skip_markers(cfg.data_root) 

75 

76 return ResetResult( 

77 deleted_docs=deleted_docs, 

78 deleted_data=deleted_data, 

79 skipped=skipped, 

80 documents_dir=str(cfg.documents_dir), 

81 data_dir=str(cfg.data_dir), 

82 )