Coverage for adhoc-cicd-odoo-odoo / odoo / orm / table_objects.py: 80%

1from __future__ import annotations 

2 

3import typing 

4 

5from odoo.tools import sql 

6 

7if typing.TYPE_CHECKING: 

8 from collections.abc import Callable 

9 

10 import psycopg2.extensions 

11 

12 from .environments import Environment 

13 from .models import BaseModel 

14 from .registry import Registry 

15 

16 ConstraintMessageType = ( 

17 str 

18 | Callable[[Environment, psycopg2.extensions.Diagnostics | None], str] 

19 ) 

20 IndexDefinitionType = ( 

21 str 

22 | Callable[[Registry], str] 

23 ) 

24 

25 

26class TableObject: 

27 """ Declares a SQL object related to the model. 

28 

29 The identifier of the SQL object will be "{model._table}_{name}". 

30 """ 

31 name: str 

32 message: ConstraintMessageType = '' 

33 _module: str = '' 

34 

35 def __init__(self): 

36 """Abstract SQL object""" 

37 # to avoid confusion: name is unique inside the model, full_name is in the database 

38 self.name = '' 

39 

40 def __set_name__(self, owner, name): 

41 # database objects should be private member fo the class: 

42 # first of all, you should not need to access them from any model 

43 # and this avoid having them in the middle of the fields when listing members 

44 assert name.startswith('_'), "Names of SQL objects in a model must start with '_'" 

45 assert not name.startswith(f"_{owner.__name__}__"), "Names of SQL objects must not be mangled" 

46 self.name = name[1:] 

47 if getattr(owner, 'pool', None) is None: # models.is_model_definition(owner) 47 ↛ exitline 47 didn't return from function '__set_name__' because the condition on line 47 was always true

48 # only for fields on definition classes, not registry classes 

49 self._module = owner._module 

50 owner._table_object_definitions.append(self) 

51 

52 def get_definition(self, registry: Registry) -> str: 

53 raise NotImplementedError 

54 

55 def full_name(self, model: BaseModel) -> str: 

56 assert self.name, f"The table object is not named ({self.definition})" 

57 name = f"{model._table}_{self.name}" 

58 return sql.make_identifier(name) 

59 

60 def get_error_message(self, model: BaseModel, diagnostics=None) -> str: 

61 """Build an error message for the object/constraint. 

62 

63 :param model: Optional model on which the constraint is defined 

64 :param diagnostics: Optional diagnostics from the raised exception 

65 :return: Translated error for the user 

66 """ 

67 message = self.message 

68 if callable(message): 

69 return message(model.env, diagnostics) 

70 return message 

71 

72 def apply_to_database(self, model: BaseModel): 

73 raise NotImplementedError 

74 

75 def __str__(self) -> str: 

76 return f"({self.name!r}={self.definition!r}, {self.message!r})" 

77 

78 

79class Constraint(TableObject): 

80 """ SQL table constraint. 

81 

82 The definition of the constraint is used to `ADD CONSTRAINT` on the table. 

83 """ 

84 

85 def __init__( 

86 self, 

87 definition: str, 

88 message: ConstraintMessageType = '', 

89 ) -> None: 

90 """ SQL table containt. 

91 

92 The definition is the SQL that will be used to add the constraint. 

93 If the constraint is violated, we will show the message to the user 

94 or an empty string to get a default message. 

95 

96 Examples of constraint definitions: 

97 - CHECK (x > 0) 

98 - FOREIGN KEY (abc) REFERENCES some_table(id) 

99 - UNIQUE (user_id) 

100 """ 

101 super().__init__() 

102 self._definition = definition 

103 if message: 103 ↛ exitline 103 didn't return from function '__init__' because the condition on line 103 was always true

104 self.message = message 

105 

106 def get_definition(self, registry: Registry): 

107 return self._definition 

108 

109 def apply_to_database(self, model: BaseModel): 

110 cr = model.env.cr 

111 conname = self.full_name(model) 

112 definition = self.get_definition(model.pool) 

113 current_definition = sql.constraint_definition(cr, model._table, conname) 

114 if current_definition == definition: 

115 return 

116 

117 if current_definition: 117 ↛ 119line 117 didn't jump to line 119 because the condition on line 117 was never true

118 # constraint exists but its definition may have changed 

119 sql.drop_constraint(cr, model._table, conname) 

120 

121 model.pool.post_constraint( 

122 cr, lambda cr: sql.add_constraint(cr, model._table, conname, definition), conname) 

123 

124 

125class Index(TableObject): 

126 """ Index on the table. 

127 

128 ``CREATE INDEX ... ON model_table <your definition>``. 

129 """ 

130 unique: bool = False 

131 

132 def __init__(self, definition: IndexDefinitionType): 

133 """ Index in SQL. 

134 

135 The name of the SQL object will be "{model._table}_{key}". The definition 

136 is the SQL that will be used to create the constraint. 

137 

138 Example of definition: 

139 - (group_id, active) WHERE active IS TRUE 

140 - USING btree (group_id, user_id) 

141 """ 

142 super().__init__() 

143 self._index_definition = definition 

144 

145 def get_definition(self, registry: Registry): 

146 if callable(self._index_definition): 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 definition = self._index_definition(registry) 

148 else: 

149 definition = self._index_definition 

150 if not definition: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 return '' 

152 return f"{'UNIQUE ' if self.unique else ''}INDEX {definition}" 

153 

154 def apply_to_database(self, model: BaseModel): 

155 cr = model.env.cr 

156 conname = self.full_name(model) 

157 definition = self.get_definition(model.pool) 

158 db_definition, db_comment = sql.index_definition(cr, conname) 

159 if db_comment == definition or (not db_comment and db_definition): 

160 # keep when the definition matches the comment in the database 

161 # or if we have an index without a comment (this is used by support to tweak indexes) 

162 return 

163 

164 if db_definition: 

165 # constraint exists but its definition may have changed 

166 sql.drop_index(cr, conname, model._table) 

167 

168 if callable(self._index_definition): 168 ↛ 169line 168 didn't jump to line 169 because the condition on line 168 was never true

169 definition_clause = self._index_definition(model.pool) 

170 else: 

171 definition_clause = self._index_definition 

172 if not definition_clause: 172 ↛ 174line 172 didn't jump to line 174 because the condition on line 172 was never true

173 # Don't create index with an empty definition 

174 return 

175 model.pool.post_constraint(cr, lambda cr: sql.add_index( 

176 cr, 

177 conname, 

178 model._table, 

179 comment=definition, 

180 definition=definition_clause, 

181 unique=self.unique, 

182 ), conname) 

183 

184 

185class UniqueIndex(Index): 

186 """ Unique index on the table. 

187 

188 ``CREATE UNIQUE INDEX ... ON model_table <your definition>``. 

189 """ 

190 unique = True 

191 

192 def __init__(self, definition: IndexDefinitionType, message: ConstraintMessageType = ''): 

193 """ Unique index in SQL. 

194 

195 The name of the SQL object will be "{model._table}_{key}". The definition 

196 is the SQL that will be used to create the constraint. 

197 You can also specify a message to be used when constraint is violated. 

198 

199 Example of definition: 

200 - (group_id, active) WHERE active IS TRUE 

201 - USING btree (group_id, user_id) 

202 """ 

203 super().__init__(definition) 

204 if message: 

205 self.message = message