Coverage for src/lektor_ng/sourcesearch.py: 90%

1import sqlite3 

2from collections.abc import Sized 

3 

4from lektor_ng.constants import PRIMARY_ALT 

5 

6 

7def _iter_parents(path): 

8 path = path.strip("/") 

9 if path: 

10 pieces = path.split("/") 

11 for x in range(len(pieces)): 

12 yield "/" + "/".join(pieces[:x]) 

13 

14 

15def _find_info(infos, alt, lang): 

16 for info in infos: 

17 if info["alt"] == alt and info["lang"] == lang: 

18 return info 

19 return None 

20 

21 

22def _id_from_path(path): 

23 try: 

24 return path.strip("/").split("/")[-1] 

25 except IndexError: 

26 return "" 

27 

28 

29def _mapping_from_cursor(cur): 

30 rv = {} 

31 for path, alt, lang, type, title in cur.fetchall(): 

32 rv.setdefault(path, []).append( 

33 { 

34 "id": _id_from_path(path), 

35 "path": path, 

36 "alt": alt, 

37 "type": type, 

38 "lang": lang, 

39 "title": title, 

40 } 

41 ) 

42 return rv 

43 

44 

45def _find_best_info(infos, alt, lang): 

46 for _alt, _lang in [ 

47 (alt, lang), 

48 (PRIMARY_ALT, lang), 

49 (alt, "en"), 

50 (PRIMARY_ALT, "en"), 

51 ]: 

52 rv = _find_info(infos, _alt, _lang) 

53 if rv is not None: 

54 return rv 

55 return None 

56 

57 

58def _build_parent_path(path, mapping, alt, lang): 

59 rv = [] 

60 for parent in _iter_parents(path): 

61 info = _find_best_info(mapping.get(parent) or [], alt, lang) 

62 id = _id_from_path(parent) 

63 if info is None: 

64 title = id or "(Index)" 

65 else: 

66 title = info.get("title") 

67 rv.append({"id": id, "path": parent, "title": title}) 

68 return rv 

69 

70 

71def _placeholders(values: Sized) -> str: 

72 """Return SQL '?' placeholders for an array or set of values.""" 

73 return ",".join(["?"] * len(values)) 

74 

75 

76def _process_search_results(builder, cur, alt, lang, limit): 

77 mapping = _mapping_from_cursor(cur) 

78 rv = [] 

79 

80 files_needed = set() 

81 

82 for path, infos in mapping.items(): 

83 info = _find_best_info(infos, alt, lang) 

84 if info is None: 

85 continue 

86 

87 for parent in _iter_parents(path): 

88 if parent not in mapping: 

89 files_needed.add(parent) 

90 

91 rv.append(info) 

92 if len(rv) == limit: 

93 break 

94 

95 if files_needed: 

96 cur.execute( 

97 f""" 

98 select path, alt, lang, type, title 

99 from source_info 

100 where path in ({_placeholders(files_needed)}) 

101 """, 

102 list(files_needed), 

103 ) 

104 mapping.update(_mapping_from_cursor(cur)) 

105 

106 for info in rv: 

107 info["parents"] = _build_parent_path(info["path"], mapping, alt, lang) 

108 

109 return rv 

110 

111 

112def find_files(builder, query, alt=PRIMARY_ALT, lang=None, limit=50, types=None): 

113 if types is None: 

114 types = ["page"] 

115 else: 

116 types = list(types) 

117 languages = ["en"] 

118 if lang not in ("en", None): 

119 languages.append(lang) 

120 else: 

121 lang = "en" 

122 alts = [PRIMARY_ALT] 

123 if alt != PRIMARY_ALT: 

124 alts.append(alt) 

125 

126 query = query.strip() 

127 title_like = "%" + query + "%" 

128 path_like = "/%" + query.rstrip("/") + "%" 

129 

130 con = sqlite3.connect(builder.buildstate_database_filename, timeout=10) 

131 try: 

132 cur = con.cursor() 

133 cur.execute( 

134 f""" 

135 select path, alt, lang, type, title 

136 from source_info 

137 where (title like ? or path like ?) 

138 and lang in ({_placeholders(languages)}) 

139 and alt in ({_placeholders(alts)}) 

140 and type in ({_placeholders(types)}) 

141 order by title 

142 collate nocase 

143 limit ? 

144 """, 

145 [title_like, path_like] + languages + alts + types + [limit * 2], 

146 ) 

147 return _process_search_results(builder, cur, alt, lang, limit) 

148 finally: 

149 con.close()