Coverage for src/lilbee/data/store/locations.py: 100%

1"""Where a source is, and the name a person sees for it: the deepest registered root's label.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Mapping 

6from pathlib import PurePath 

7 

8from .lance_helpers import escape_sql_string 

9from .types import SourceType 

10 

11 

12def root_location(root: str) -> str: 

13 """The location of the registered root path *root*: its POSIX spelling.""" 

14 return PurePath(root).as_posix() 

15 

16 

17def _escape_like_wildcards(value: str) -> str: 

18 """*value* with the LIKE wildcards and the escape character escaped, matched literally.""" 

19 return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") 

20 

21 

22def under_sql(column: str, root: str) -> str: 

23 """The predicate for *column* holding *root* or a key below it, in parentheses. 

24 

25 ``starts_with`` takes the key literally, so no character in a key is a wildcard. 

26 """ 

27 literal = escape_sql_string(root) 

28 return f"({column} = '{literal}' OR starts_with({column}, '{literal}/'))" 

29 

30 

31class SourceLocations: 

32 """The rule between a stored location and its shown name under one source registry. 

33 

34 A location below a registered root shows the label of the deepest such root 

35 plus the rest of the path; any other location shows itself. 

36 """ 

37 

38 def __init__(self, roots: Mapping[str, str]) -> None: 

39 self.roots = dict(roots) 

40 self._at = {label: root_location(root) for label, root in roots.items()} 

41 self._deepest_first = sorted( 

42 ((root, label) for label, root in self._at.items()), 

43 key=lambda pair: len(pair[0]), 

44 reverse=True, 

45 ) 

46 

47 @property 

48 def deepest_first(self) -> list[tuple[str, str]]: 

49 """Each registered root's location and label, a root below another before it.""" 

50 return list(self._deepest_first) 

51 

52 def name(self, location: str) -> str: 

53 """The name a person sees for the source at *location*.""" 

54 for root, label in self._deepest_first: 

55 if location == root or location.startswith(f"{root}/"): 

56 return label + location[len(root) :] 

57 return location 

58 

59 def location(self, name: str) -> str: 

60 """Where the source named *name* is: its root's location plus the rest, else *name*.""" 

61 label, _, rest = name.partition("/") 

62 root = self._at.get(label) 

63 if root is None: 

64 return name 

65 return f"{root}/{rest}" if rest else root 

66 

67 def key(self, name: str, source_type: SourceType) -> str: 

68 """What ``_sources.path`` holds for *name*: a document's location, else the name.""" 

69 return self.location(name) if source_type == SourceType.DOCUMENT else name 

70 

71 def named_by_sql(self, column: str, root: str) -> str: 

72 """The predicate for *column* holding a location whose name the registered *root* gives.""" 

73 deeper = [ 

74 f"NOT {under_sql(column, below)}" 

75 for below, _ in self._deepest_first 

76 if below.startswith(f"{root}/") 

77 ] 

78 return " AND ".join([under_sql(column, root), *deeper]) 

79 

80 def search_filter(self, term: str, path: str, title: str) -> str: 

81 """The predicate for a shown name in *path*, or a title in *title*, holding *term*. 

82 

83 Case-insensitive. Each root's arm leaves out the roots below it, so a 

84 location matches by the one name it shows. 

85 """ 

86 pattern = f"LIKE '%{escape_sql_string(_escape_like_wildcards(term.lower()))}%' ESCAPE '\\'" 

87 arms: list[str] = [] 

88 for root, label in self._deepest_first: 

89 shown = f"concat('{escape_sql_string(label)}', substr({path}, {len(root) + 1}))" 

90 arms.append(f"{self.named_by_sql(path, root)} AND LOWER({shown}) {pattern}") 

91 outside = [f"NOT {under_sql(path, root)}" for root, _ in self._deepest_first] 

92 arms.append(" AND ".join([*outside, f"LOWER({path}) {pattern}"])) 

93 arms.append(f"LOWER({title}) {pattern}") 

94 return "(" + " OR ".join(f"({arm})" for arm in arms) + ")"