Coverage for app/venv/lib/python3.14/site-packages/weblate/utils/db.py: 35%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5"""Database specific code to extend Django.""" 

6 

7from __future__ import annotations 

8 

9import time 

10 

11from django.db import ProgrammingError, connections, models, transaction 

12from django.db.models.lookups import PatternLookup, Regex 

13 

14from .inv_regex import invert_re 

15 

16ESCAPED = frozenset(".\\+*?[^]$(){}=!<>|:-") 

17 

18PG_TRGM = "CREATE INDEX {0}_{1}_fulltext ON trans_{0} USING GIN ({1} gin_trgm_ops {2})" 

19PG_DROP = "DROP INDEX {0}_{1}_fulltext" 

20 

21MY_FTX = "CREATE FULLTEXT INDEX {0}_{1}_fulltext ON trans_{0}({1})" 

22MY_DROP = "ALTER TABLE trans_{0} DROP INDEX {0}_{1}_fulltext" 

23 

24 

25class MissingTransactionError(ProgrammingError): 

26 pass 

27 

28 

29def using_postgresql(): 

30 return connections["default"].vendor == "postgresql" 

31 

32 

33class TransactionsTestMixin: 

34 @classmethod 

35 def _databases_support_transactions(cls): 

36 # This is workaround for MySQL as FULL TEXT index does not work 

37 # well inside a transaction, so we avoid using transactions for 

38 # tests. Otherwise we end up with no matches for the query. 

39 # See https://dev.mysql.com/doc/refman/5.6/en/innodb-fulltext-index.html 

40 if not using_postgresql(): 

41 return False 

42 return super()._databases_support_transactions() # type: ignore[misc] 

43 

44 

45def adjust_similarity_threshold(value: float) -> None: 

46 """ 

47 Adjust pg_trgm.similarity_threshold for the % operator. 

48 

49 Ideally we would use directly similarity() in the search, but that doesn't seem 

50 to use index, while using % does. 

51 """ 

52 if not using_postgresql(): 

53 return 

54 

55 if "memory_db" in connections: 

56 connection = connections["memory_db"] 

57 else: 

58 connection = connections["default"] 

59 

60 current_similarity = getattr(connection, "weblate_similarity", -1) 

61 # Ignore small differences 

62 if abs(current_similarity - value) < 0.05: 

63 return 

64 

65 with connection.cursor() as cursor: 

66 # The SELECT has to be executed first as otherwise the trgm extension 

67 # might not yet be loaded and GUC setting not possible. 

68 if current_similarity == -1: 

69 cursor.execute("SELECT show_limit()") 

70 

71 # Adjust threshold 

72 cursor.execute("SELECT set_limit(%s)", [value]) 

73 connection.weblate_similarity = value # type: ignore[attr-defined] 

74 

75 

76def count_alnum(string): 

77 return sum(map(str.isalnum, string)) 

78 

79 

80class PostgreSQLFallbackLookupMixin: 

81 """ 

82 Mixin to block PostgreSQL from using trigram index. 

83 

84 It is ineffective for very short strings as these produce a lot of matches 

85 which need to be rechecked and full table scan is more effective in that 

86 case. 

87 

88 It is performed by concatenating empty string which will prevent index usage. 

89 """ 

90 

91 def process_lhs(self, compiler, connection, lhs=None): 

92 if self._needs_fallback: # type: ignore[attr-defined] 

93 lhs_sql, params = super().process_lhs(compiler, connection, lhs) # type: ignore[misc] 

94 if self.lookup_name in {"search", "substring"}: 

95 # These are matched against UPPER, so convert them 

96 return f"UPPER({lhs_sql})", params 

97 # This concatenation will prevent using trigram index 

98 return f"{lhs_sql} || ''", params 

99 return super().process_lhs(compiler, connection, lhs) # type: ignore[misc] 

100 

101 

102class PostgreSQLFallbackLookup(PostgreSQLFallbackLookupMixin, PatternLookup): 

103 def __init__(self, lhs, rhs) -> None: 

104 self._needs_fallback = isinstance(rhs, str) and count_alnum(rhs) <= 3 

105 super().__init__(lhs, rhs) 

106 

107 

108class PostgreSQLRegexLookup(PostgreSQLFallbackLookupMixin, Regex): 

109 def __init__(self, lhs, rhs) -> None: 

110 self._needs_fallback = isinstance(rhs, str) and ( 

111 min((count_alnum(match) for match in invert_re(rhs)), default=0) < 3 

112 ) 

113 super().__init__(lhs, rhs) 

114 

115 

116class PostgreSQLSearchLookup(PostgreSQLFallbackLookup): 

117 lookup_name = "search" 

118 

119 def process_rhs(self, qn, connection): 

120 if not self._needs_fallback: 

121 self.param_pattern = "%s" 

122 return super().process_rhs(qn, connection) 

123 

124 def get_rhs_op(self, connection, rhs): 

125 if self._needs_fallback: 

126 return connection.operators["icontains"] % rhs 

127 return f"%% {rhs} = true" 

128 

129 

130class MySQLSearchLookup(models.Lookup): 

131 lookup_name = "search" 

132 

133 def as_sql(self, compiler, connection): 

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

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

136 params = lhs_params + rhs_params # type: ignore[operator] 

137 return f"MATCH ({lhs}) AGAINST ({rhs} IN NATURAL LANGUAGE MODE)", params 

138 

139 

140class PostgreSQLSubstringLookup(PostgreSQLFallbackLookup): 

141 """ 

142 Case insensitive substring lookup. 

143 

144 This is essentially same as icontains in Django, but utilizes ILIKE 

145 operator which can use pg_trgm index. 

146 """ 

147 

148 lookup_name = "substring" 

149 

150 def get_rhs_op(self, connection, rhs): 

151 if self._needs_fallback: 

152 return connection.operators["icontains"] % rhs 

153 return f"ILIKE {rhs}" 

154 

155 

156def re_escape(pattern: str) -> str: 

157 """ 

158 Escape for use in database regexp match. 

159 

160 This is based on re.escape, but that one escapes too much. 

161 """ 

162 string = list(pattern) 

163 for i, char in enumerate(pattern): 

164 if char == "\000": 

165 string[i] = "\\000" 

166 elif char in ESCAPED: 

167 string[i] = "\\" + char 

168 return "".join(string) 

169 

170 

171def measure_database_latency() -> float: 

172 from weblate.trans.models import Project 

173 

174 start = time.monotonic() 

175 Project.objects.exists() 

176 return round(1000 * (time.monotonic() - start)) 

177 

178 

179def verify_in_transaction() -> None: 

180 """Verify the code is executed inside a transaction.""" 

181 connection = transaction.get_connection() 

182 if not connection.in_atomic_block: 

183 raise MissingTransactionError