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.config import cfg 

12from lilbee.core.security import validate_path_within 

13 

14 

15class ResetResult(BaseModel): 

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

17 

18 command: str = "reset" 

19 deleted_docs: int 

20 deleted_data: int 

21 skipped: list[str] = [] 

22 documents_dir: str 

23 data_dir: str 

24 

25 

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

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

28 log = logging.getLogger(__name__) 

29 deleted = 0 

30 if not base_dir.exists(): 

31 return deleted 

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

33 validate_path_within(item, base_dir) 

34 try: 

35 if item.is_dir(): 

36 shutil.rmtree(item) 

37 else: 

38 item.unlink() 

39 except OSError as exc: 

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

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

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

43 skipped.append(str(item)) 

44 continue 

45 deleted += 1 

46 return deleted 

47 

48 

49def perform_reset() -> ResetResult: 

50 """Delete all documents and data. Returns summary of what was deleted.""" 

51 skipped: list[str] = [] 

52 deleted_docs = _clear_dir(cfg.documents_dir, skipped) 

53 deleted_data = _clear_dir(cfg.data_dir, skipped) 

54 

55 return ResetResult( 

56 deleted_docs=deleted_docs, 

57 deleted_data=deleted_data, 

58 skipped=skipped, 

59 documents_dir=str(cfg.documents_dir), 

60 data_dir=str(cfg.data_dir), 

61 )