Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/skeleton/adapter.py: 33%

1import typing as t 

2from itertools import chain 

3from .. import db 

4from ..config import conf 

5 

6 

7class DatabaseAdapter: 

8 """ 

9 Adapter class used to bind or use other databases and hook operations when working with a Skeleton. 

10 """ 

11 

12 providesFulltextSearch: bool = False 

13 """Set to True if we can run a fulltext search using this database.""" 

14 

15 fulltextSearchGuaranteesQueryConstrains = False 

16 """Are results returned by `meth:fulltextSearch` guaranteed to also match the databaseQuery""" 

17 

18 providesCustomQueries: bool = False 

19 """Indicate that we can run more types of queries than originally supported by datastore""" 

20 

21 def prewrite(self, skel: "SkeletonInstance", is_add: bool, change_list: t.Iterable[str] = ()): 

22 """ 

23 Hook being called on a add, edit or delete operation before the skeleton-specific action is performed. 

24 

25 The hook can be used to modifiy the skeleton before writing. 

26 The raw entity can be obainted using `skel.dbEntity`. 

27 

28 :param action: Either contains "add", "edit" or "delete", depending on the operation. 

29 :param skel: is the skeleton that is being read before written. 

30 :param change_list: is a list of bone names which are being changed within the write. 

31 """ 

32 pass 

33 

34 def write(self, skel: "SkeletonInstance", is_add: bool, change_list: t.Iterable[str] = ()): 

35 """ 

36 Hook being called on a write operations after the skeleton is written. 

37 

38 The raw entity can be obainted using `skel.dbEntity`. 

39 

40 :param action: Either contains "add" or "edit", depending on the operation. 

41 :param skel: is the skeleton that is being read before written. 

42 :param change_list: is a list of bone names which are being changed within the write. 

43 """ 

44 pass 

45 

46 def delete(self, skel: "SkeletonInstance"): 

47 """ 

48 Hook being called on a delete operation after the skeleton is deleted. 

49 """ 

50 pass 

51 

52 def fulltextSearch(self, queryString: str, databaseQuery: db.Query) -> list[db.Entity]: 

53 """ 

54 If this database supports fulltext searches, this method has to implement them. 

55 If it's a plain fulltext search engine, leave 'prop:fulltextSearchGuaranteesQueryConstrains' set to False, 

56 then the server will post-process the list of entries returned from this function and drop any entry that 

57 cannot be returned due to other constrains set in 'param:databaseQuery'. If you can obey *every* constrain 

58 set in that Query, we can skip this post-processing and save some CPU-cycles. 

59 :param queryString: the string as received from the user (no quotation or other safety checks applied!) 

60 :param databaseQuery: The query containing any constrains that returned entries must also match 

61 :return: 

62 """ 

63 raise NotImplementedError 

64 

65 

66class ViurTagsSearchAdapter(DatabaseAdapter): 

67 """ 

68 This Adapter implements a simple fulltext search on top of the datastore. 

69 

70 On skel.write(), all words from String-/TextBones are collected with all *min_length* postfixes and dumped 

71 into the property `viurTags`. When queried, we'll run a prefix-match against this property - thus returning 

72 entities with either an exact match or a match within a word. 

73 

74 Example: 

75 For the word "hello" we'll write "hello", "ello" and "llo" into viurTags. 

76 When queried with "hello" we'll have an exact match. 

77 When queried with "hel" we'll match the prefix for "hello" 

78 When queried with "ell" we'll prefix-match "ello" - this is only enabled when substring_matching is True. 

79 

80 We'll automatically add this adapter if a skeleton has no other database adapter defined. 

81 """ 

82 providesFulltextSearch = True 

83 fulltextSearchGuaranteesQueryConstrains = True 

84 

85 def __init__(self, min_length: int = 2, max_length: int = 50, substring_matching: bool = False): 

86 super().__init__() 

87 self.min_length = min_length 

88 self.max_length = max_length 

89 self.substring_matching = substring_matching 

90 

91 def _tags_from_str(self, value: str) -> set[str]: 

92 """ 

93 Extract all words including all min_length postfixes from given string 

94 """ 

95 res = set() 

96 

97 for tag in value.split(" "): 

98 tag = "".join([x for x in tag.lower() if x in conf.search_valid_chars]) 

99 

100 if len(tag) >= self.min_length: 

101 res.add(tag) 

102 

103 if self.substring_matching: 

104 for i in range(1, 1 + len(tag) - self.min_length): 

105 res.add(tag[i:]) 

106 

107 return res 

108 

109 def prewrite(self, skel: "SkeletonInstance", *args, **kwargs): 

110 """ 

111 Collect searchTags from skeleton and build viurTags 

112 """ 

113 tags = set() 

114 

115 for name, bone in skel.items(): 

116 if bone.searchable: 

117 tags = tags.union(bone.getSearchTags(skel, name)) 

118 

119 skel.dbEntity["viurTags"] = list( 

120 chain(*[self._tags_from_str(tag) for tag in tags if len(tag) <= self.max_length]) 

121 ) 

122 

123 def fulltextSearch(self, queryString: str, databaseQuery: db.Query) -> list[db.Entity]: 

124 """ 

125 Run a fulltext search 

126 """ 

127 keywords = list(self._tags_from_str(queryString))[:10] 

128 resultScoreMap = {} 

129 resultEntryMap = {} 

130 

131 for keyword in keywords: 

132 qryBase = databaseQuery.clone() 

133 for entry in qryBase.filter("viurTags >=", keyword).filter("viurTags <", keyword + "\ufffd").run(): 

134 if entry.key not in resultScoreMap: 

135 resultScoreMap[entry.key] = 1 

136 else: 

137 resultScoreMap[entry.key] += 1 

138 if entry.key not in resultEntryMap: 

139 resultEntryMap[entry.key] = entry 

140 

141 resultList = [(k, v) for k, v in resultScoreMap.items()] 

142 resultList.sort(key=lambda x: x[1], reverse=True) 

143 

144 return [resultEntryMap[x[0]] for x in resultList[:databaseQuery.queries.limit]]