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 

12 

13 

14class ResetResult(BaseModel): 

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

16 

17 command: str = "reset" 

18 deleted_docs: int 

19 deleted_data: int 

20 skipped: list[str] = [] 

21 documents_dir: str 

22 data_dir: str 

23 

24 

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

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

27 log = logging.getLogger(__name__) 

28 deleted = 0 

29 if not base_dir.exists(): 

30 return deleted 

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

32 try: 

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

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

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

36 # user's corpus: it deletes owned files, and un-registering roots is 

37 # done by clearing config.toml under data_dir. 

38 if item.is_dir(): 

39 shutil.rmtree(item) 

40 else: 

41 item.unlink() 

42 except OSError as exc: 

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

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

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

46 skipped.append(str(item)) 

47 continue 

48 deleted += 1 

49 return deleted 

50 

51 

52def perform_reset() -> ResetResult: 

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

54 skipped: list[str] = [] 

55 deleted_docs = _clear_dir(cfg.documents_dir, skipped) 

56 deleted_data = _clear_dir(cfg.data_dir, skipped) 

57 

58 return ResetResult( 

59 deleted_docs=deleted_docs, 

60 deleted_data=deleted_data, 

61 skipped=skipped, 

62 documents_dir=str(cfg.documents_dir), 

63 data_dir=str(cfg.data_dir), 

64 )