Coverage for apis_core/search/models.py: 45%

1import unicodedata 

2 

3from django.conf import settings 

4from django.contrib.contenttypes.fields import GenericForeignKey 

5from django.contrib.contenttypes.models import ContentType 

6from django.contrib.postgres.indexes import GinIndex 

7from django.contrib.postgres.search import ( 

8 SearchQuery, 

9 SearchRank, 

10 SearchVector, 

11 SearchVectorField, 

12) 

13from django.db import models 

14from django.db.models import Case, F, Field, Q, Value, When 

15from django.db.models.functions import Greatest 

16from django.db.models.lookups import PatternLookup 

17 

18from .utils import ( 

19 get_search_serialization_from_instance, 

20 split_query_string_for_search, 

21) 

22 

23 

24@Field.register_lookup 

25class ILike(PatternLookup): 

26 """ 

27 We implement our own `ilike` lookup, because the `icontains` 

28 lookup uses the sql `UPPER` function and matching functions 

29 with a lookup index does not work. 

30 """ 

31 

32 lookup_name = "ilike" 

33 

34 def as_sql(self, compiler, connection): 

35 lhs, lhs_params = self.process_lhs(compiler, connection) 

36 rhs, rhs_params = self.process_rhs(compiler, connection) 

37 params = (*lhs_params, *rhs_params) 

38 return "%s::text ILIKE %s" % (lhs, rhs), params 

39 

40 

41class SearchManager(models.Manager): 

42 def search(self, query_str: str, content_types=[], with_content=False): 

43 """ 

44 We combine multiple approaches to searching the entries. 

45 We use the `search_vector` full text search on the title 

46 and the content_text fields. Any calculated rank that is 

47 lower than 0.01 is set to 0. We then store the maximum 

48 value of the title rank times 2 and the content rank. This 

49 is then used the primary means for ranking. 

50 Then we look at the individual fields and give them weight 

51 in the order title -> content_text, each both case-sensitive 

52 & case insensitive. We match those fields for the whole 

53 `query_str` as well as the individual parts. 

54 Those checks are then also used as an overall filter, so only results 

55 containing those strings are looked at and returned. 

56 """ 

57 

58 query = super().get_queryset() 

59 search_config = getattr(settings, "SEARCH_SEARCHQUERY_CONFIG", None) 

60 search_query = SearchQuery( 

61 unicodedata.normalize("NFKD", query_str), config=search_config 

62 ) 

63 query = query.annotate( 

64 title_rank=SearchRank(F("title_search_vector"), search_query), 

65 content_rank=Value(0.0), 

66 ) 

67 if with_content: 

68 query = query.annotate( 

69 content_rank=SearchRank(F("content_search_vector"), search_query) 

70 ) 

71 query = query.annotate( 

72 title_rank=Case( 

73 When(title_rank__gt=0.01, then=F("title_rank")), default=Value(0.0) 

74 ) 

75 ) 

76 query = query.annotate( 

77 content_rank=Case( 

78 When(content_rank__gt=0.01, then=F("content_rank")), default=Value(0.0) 

79 ) 

80 ) 

81 query = query.annotate(frank=Greatest(F("title_rank") * 2, "content_rank")) 

82 

83 lookups = { 

84 "title__contains": 10, 

85 "title__ilike": 9, 

86 } 

87 if with_content: 

88 lookups["content_text__contains"] = 6 

89 lookups["content_text__ilike"] = 5 

90 

91 parts = split_query_string_for_search(query_str) 

92 q = Q() 

93 w = [] 

94 

95 weights = [0] * len(parts) 

96 weights[0] = 2 

97 for weight, part in zip(weights, parts): 

98 for lookup, value in lookups.items(): 

99 q |= Q(**{lookup: part}) 

100 w += [When(**{lookup: part, "then": value + weight})] 

101 

102 query = query.filter(q).annotate(contains=Case(*w, default=0)) 

103 

104 query = query.order_by("-frank", "-contains", "id") 

105 

106 if content_types: 

107 query = query.filter(content_type__in=content_types) 

108 return query 

109 

110 

111class SearchEntry(models.Model): 

112 """ 

113 An entry in the search table pointing to a Django model instance 

114 """ 

115 

116 content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, null=True) 

117 object_id = models.PositiveIntegerField(null=True) 

118 content_object = GenericForeignKey() 

119 

120 title = models.CharField(max_length=1024) 

121 

122 content = models.JSONField() 

123 

124 content_text = models.TextField(null=True) 

125 

126 title_search_vector = SearchVectorField() 

127 content_search_vector = SearchVectorField() 

128 

129 objects = SearchManager() 

130 

131 class Meta: 

132 app_label = "search" 

133 unique_together = ["content_type", "object_id"] 

134 indexes = [ 

135 models.Index(fields=["content_type", "object_id"]), 

136 GinIndex( 

137 name="generic_sea_content_text_gin", 

138 fields=["content_text"], 

139 opclasses=["gin_trgm_ops"], 

140 ), 

141 GinIndex( 

142 name="generic_sea_title_gin", 

143 fields=["title"], 

144 opclasses=["gin_trgm_ops"], 

145 ), 

146 GinIndex(fields=["title_search_vector"]), 

147 GinIndex(fields=["content_search_vector"]), 

148 ] 

149 

150 @classmethod 

151 def reindex_model_instance(cls, instance): 

152 content_type = ContentType.objects.get_for_model(instance) 

153 

154 title = unicodedata.normalize("NFKD", str(instance)) 

155 

156 serialization = get_search_serialization_from_instance(instance) 

157 

158 serialization_text = {k: str(v) for k, v in serialization.items() if v} 

159 content_text = " ".join(serialization_text.values()) 

160 

161 defaults = { 

162 "title": title, 

163 "content": serialization, 

164 "content_text": content_text, 

165 } 

166 s, c = cls.objects.update_or_create( 

167 content_type=content_type, object_id=instance.pk, defaults=defaults 

168 ) 

169 s.title_search_vector = SearchVector("title", weight="A") 

170 s.content_search_vector = SearchVector("content_text", weight="A") 

171 s.save() 

172 

173 @classmethod 

174 def delete_by_instance(cls, instance): 

175 content_type = ContentType.objects.get_for_model(instance) 

176 

177 cls.objects.filter(content_type=content_type, object_id=instance.pk).delete()