Coverage for adhoc-cicd-odoo-odoo / odoo / orm / fields_relational.py: 75%

1from __future__ import annotations 

2 

3import itertools 

4import logging 

5import typing 

6from collections import defaultdict 

7from collections.abc import Reversible 

8from operator import attrgetter 

9 

10from odoo.exceptions import AccessError, MissingError, UserError 

11from odoo.tools import SQL, OrderedSet, Query, sql, unique 

12from odoo.tools.constants import PREFETCH_MAX 

13from odoo.tools.misc import SENTINEL, Sentinel, unquote 

14 

15from .commands import Command 

16from .domains import Domain 

17from .fields import IR_MODELS, Field, _logger 

18from .fields_reference import Many2oneReference 

19from .identifiers import NewId 

20from .models import BaseModel 

21from .utils import COLLECTION_TYPES, SQL_OPERATORS, check_pg_name 

22 

23if typing.TYPE_CHECKING: 

24 from collections.abc import Sequence 

25 from odoo.tools.misc import Collector 

26 from .types import CommandValue, ContextType, DomainType, Environment, Registry 

27 

28 OnDelete = typing.Literal['cascade', 'set null', 'restrict'] 

29 

30_schema = logging.getLogger('odoo.schema') 

31 

32 

33class _Relational(Field[BaseModel]): 

34 """ Abstract class for relational fields. """ 

35 relational: typing.Literal[True] = True 

36 comodel_name: str 

37 domain: DomainType = [] # domain for searching values 

38 context: ContextType = {} # context for searching values 

39 bypass_search_access: bool = False # whether access rights are bypassed on the comodel 

40 check_company: bool = False 

41 

42 def __get__(self, records: BaseModel, owner=None): 

43 # base case: do the regular access 

44 if records is None or len(records._ids) <= 1: 

45 return super().__get__(records, owner) 

46 

47 records._check_field_access(self, 'read') 

48 

49 # multi-record case 

50 if self.compute and self.store: 

51 self.recompute(records) 

52 

53 # get the cache 

54 env = records.env 

55 field_cache = self._get_cache(env) 

56 

57 # retrieve values in cache, and fetch missing ones 

58 vals = [] 

59 for record_id in records._ids: 

60 try: 

61 vals.append(field_cache[record_id]) 

62 except KeyError: 

63 if self.store and record_id and len(vals) < len(records) - PREFETCH_MAX: 63 ↛ 65line 63 didn't jump to line 65 because the condition on line 63 was never true

64 # a lot of missing records, just fetch that field 

65 remaining = records[len(vals):] 

66 remaining.fetch([self.name]) 

67 # fetch does not raise MissingError, check value 

68 if record_id not in field_cache: 

69 raise MissingError("\n".join([ 

70 env._("Record does not exist or has been deleted."), 

71 env._("(Record: %(record)s, User: %(user)s)", record=record_id, user=env.uid), 

72 ])) from None 

73 else: 

74 remaining = records.__class__(env, (record_id,), records._prefetch_ids) 

75 super().__get__(remaining, owner) 

76 # we have the record now 

77 vals.append(field_cache[record_id]) 

78 

79 return self.convert_to_record_multi(vals, records) 

80 

81 def _update_inverse(self, records: BaseModel, value: BaseModel): 

82 """ Update the cached value of ``self`` for ``records`` with ``value``. """ 

83 raise NotImplementedError 

84 

85 def convert_to_record_multi(self, values, records): 

86 """ Convert a list of (relational field) values from the cache format to 

87 the record format, for the sake of optimization. 

88 """ 

89 raise NotImplementedError 

90 

91 def setup_nonrelated(self, model): 

92 super().setup_nonrelated(model) 

93 assert self.comodel_name in model.pool, \ 

94 f"Field {self} with unknown comodel_name {self.comodel_name or '???'!r}" 

95 

96 def setup_inverses(self, registry: Registry, inverses: Collector[Field, Field]): 

97 """ Populate ``inverses`` with ``self`` and its inverse fields. """ 

98 

99 def get_comodel_domain(self, model: BaseModel) -> Domain: 

100 """ Return a domain from the domain attribute. """ 

101 domain = self.domain 

102 if callable(domain): 

103 # the callable can return either a list, Domain or a string 

104 domain = domain(model) 

105 if not domain or isinstance(domain, str): 

106 # if we don't have a domain or 

107 # domain=str is used only for the client-side 

108 return Domain.TRUE 

109 return Domain(domain) 

110 

111 @property 

112 def _related_domain(self) -> DomainType | None: 

113 def validated(domain): 

114 if isinstance(domain, str) and not self.inherited: 

115 # string domains are expressions that are not valid for self's model 

116 return None 

117 return domain 

118 

119 if callable(self.domain): 

120 # will be called with another model than self's 

121 return lambda recs: validated(self.domain(recs.env[self.model_name])) # pylint: disable=not-callable 

122 else: 

123 return validated(self.domain) 

124 

125 _related_context = property(attrgetter('context')) 

126 

127 _description_relation = property(attrgetter('comodel_name')) 

128 _description_context = property(attrgetter('context')) 

129 

130 def _description_domain(self, env: Environment) -> str | list: 

131 domain = self._internal_description_domain_raw(env) 

132 if self.check_company: 

133 field_to_check = None 

134 if self.company_dependent: 

135 cids = '[allowed_company_ids[0]]' 

136 elif self.model_name == 'res.company': 

137 # when using check_company=True on a field on 'res.company', the 

138 # company_id comes from the id of the current record 

139 cids = '[id]' 

140 elif 'company_id' in env[self.model_name]: 

141 cids = '[company_id]' 

142 field_to_check = 'company_id' 

143 elif 'company_ids' in env[self.model_name]: 143 ↛ 147line 143 didn't jump to line 147 because the condition on line 143 was always true

144 cids = 'company_ids' 

145 field_to_check = 'company_ids' 

146 else: 

147 _logger.warning(env._( 

148 "Couldn't generate a company-dependent domain for field %s. " 

149 "The model doesn't have a 'company_id' or 'company_ids' field, and isn't company-dependent either.", 

150 self.model_name + '.' + self.name, 

151 )) 

152 return domain 

153 company_domain = env[self.comodel_name]._check_company_domain(companies=unquote(cids)) 

154 if not field_to_check: 

155 return f"{company_domain} + {domain or []}" 

156 else: 

157 no_company_domain = env[self.comodel_name]._check_company_domain(companies='') 

158 return f"({field_to_check} and {company_domain} or {no_company_domain}) + ({domain or []})" 

159 return domain 

160 

161 def _description_allow_hierachy_operators(self, env): 

162 """ Return if the child_of/parent_of makes sense on this field """ 

163 comodel = env[self.comodel_name] 

164 return comodel._parent_name in comodel._fields 

165 

166 def _internal_description_domain_raw(self, env) -> str | list: 

167 domain = self.domain 

168 if callable(domain): 

169 domain = domain(env[self.model_name]) 

170 if isinstance(domain, Domain): 

171 domain = list(domain) 

172 return domain or [] 

173 

174 def filter_function(self, records, field_expr, operator, value): 

175 getter = self.expression_getter(field_expr) 

176 

177 if (self.bypass_search_access or operator == 'any!') and not records.env.su: 177 ↛ 181line 177 didn't jump to line 181 because the condition on line 177 was never true

178 # When filtering with bypass access, search the corecords with sudo 

179 # and a special key in the context. To evaluate sub-domains, the 

180 # special key makes the environment un-sudoed before evaluation. 

181 expr_getter = getter 

182 sudo_env = records.sudo().with_context(filter_function_reset_sudo=True).env 

183 getter = lambda rec: expr_getter(rec.with_env(sudo_env)) # noqa: E731 

184 

185 corecords = getter(records) 

186 if operator in ('any', 'any!'): 

187 assert isinstance(value, Domain) 

188 if operator == 'any' and records.env.context.get('filter_function_reset_sudo'): 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true

189 corecords = corecords.sudo(False)._filtered_access('read') 

190 corecords = corecords.filtered_domain(value) 

191 elif operator == 'in' and isinstance(value, COLLECTION_TYPES): 191 ↛ 204line 191 didn't jump to line 204 because the condition on line 191 was always true

192 value = set(value) 

193 if False in value: 

194 if not corecords: 

195 # shortcut, we know none of records has a corecord 

196 return lambda _: True 

197 if len(value) > 1: 

198 value.discard(False) 

199 filter_values = self.filter_function(records, field_expr, 'in', value) 

200 return lambda rec: not getter(rec) or filter_values(rec) 

201 return lambda rec: not getter(rec) 

202 corecords = corecords.filtered_domain(Domain('id', 'in', value)) 

203 else: 

204 corecords = corecords.filtered_domain(Domain('id', operator, value)) 

205 

206 if not corecords: 

207 return lambda _: False 

208 

209 ids = set(corecords._ids) 

210 return lambda rec: any(id_ in ids for val in getter(rec) for id_ in val._ids) 

211 

212 

213class Many2one(_Relational): 

214 """ The value of such a field is a recordset of size 0 (no 

215 record) or 1 (a single record). 

216 

217 :param str comodel_name: name of the target model 

218 ``Mandatory`` except for related or extended fields. 

219 

220 :param domain: an optional domain to set on candidate values on the 

221 client side (domain or a python expression that will be evaluated 

222 to provide domain) 

223 

224 :param dict context: an optional context to use on the client side when 

225 handling that field 

226 

227 :param str ondelete: what to do when the referred record is deleted; 

228 possible values are: ``'set null'``, ``'restrict'``, ``'cascade'`` 

229 

230 :param bool bypass_search_access: whether access rights are bypassed on the 

231 comodel (default: ``False``) 

232 

233 :param bool delegate: set it to ``True`` to make fields of the target model 

234 accessible from the current model (corresponds to ``_inherits``) 

235 

236 :param bool check_company: Mark the field to be verified in 

237 :meth:`~odoo.models.Model._check_company`. Has a different behaviour 

238 depending on whether the field is company_dependent or not. 

239 Constrains non-company-dependent fields to target records whose 

240 company_id(s) are compatible with the record's company_id(s). 

241 Constrains company_dependent fields to target records whose 

242 company_id(s) are compatible with the currently active company. 

243 """ 

244 type = 'many2one' 

245 _column_type = ('int4', 'int4') 

246 

247 ondelete: OnDelete | None = None # what to do when value is deleted 

248 delegate: bool = False # whether self implements delegation 

249 

250 def __init__(self, comodel_name: str | Sentinel = SENTINEL, string: str | Sentinel = SENTINEL, **kwargs): 

251 super().__init__(comodel_name=comodel_name, string=string, **kwargs) 

252 

253 def _setup_attrs__(self, model_class, name): 

254 super()._setup_attrs__(model_class, name) 

255 # determine self.delegate 

256 if name in model_class._inherits.values(): 

257 self.delegate = True 

258 # self.delegate implies self.bypass_search_access 

259 self.bypass_search_access = True 

260 elif self.delegate: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true

261 comodel_name = self.comodel_name or 'comodel_name' 

262 raise TypeError(( 

263 f"The delegate field {self} must be declared in the model class e.g.\n" 

264 f"_inherits = {{{comodel_name!r}: {name!r}}}" 

265 )) 

266 

267 def setup_nonrelated(self, model): 

268 super().setup_nonrelated(model) 

269 # 3 cases: 

270 # 1) The ondelete attribute is not defined, we assign it a sensible default 

271 # 2) The ondelete attribute is defined and its definition makes sense 

272 # 3) The ondelete attribute is explicitly defined as 'set null' for a required m2o, 

273 # this is considered a programming error. 

274 if not self.ondelete: 

275 comodel = model.env[self.comodel_name] 

276 if model.is_transient() and not comodel.is_transient(): 

277 # Many2one relations from TransientModel Model are annoying because 

278 # they can block deletion due to foreign keys. So unless stated 

279 # otherwise, we default them to ondelete='cascade'. 

280 self.ondelete = 'cascade' if self.required else 'set null' 

281 else: 

282 self.ondelete = 'restrict' if self.required else 'set null' 

283 if self.ondelete == 'set null' and self.required: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true

284 raise ValueError( 

285 "The m2o field %s of model %s is required but declares its ondelete policy " 

286 "as being 'set null'. Only 'restrict' and 'cascade' make sense." 

287 % (self.name, model._name) 

288 ) 

289 if self.ondelete == 'restrict' and self.comodel_name in IR_MODELS: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true

290 raise ValueError( 

291 f"Field {self.name} of model {model._name} is defined as ondelete='restrict' " 

292 f"while having {self.comodel_name} as comodel, the 'restrict' mode is not " 

293 f"supported for this type of field as comodel." 

294 ) 

295 

296 def update_db(self, model, columns): 

297 comodel = model.env[self.comodel_name] 

298 if not model.is_transient() and comodel.is_transient(): 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true

299 raise ValueError('Many2one %s from Model to TransientModel is forbidden' % self) 

300 return super().update_db(model, columns) 

301 

302 def update_db_column(self, model, column): 

303 super().update_db_column(model, column) 

304 model.pool.post_init(self.update_db_foreign_key, model, column) 

305 

306 def update_db_foreign_key(self, model, column): 

307 if self.company_dependent: 

308 return 

309 comodel = model.env[self.comodel_name] 

310 # foreign keys do not work on views, and users can define custom models on sql views. 

311 if not model._is_an_ordinary_table() or not comodel._is_an_ordinary_table(): 311 ↛ 312line 311 didn't jump to line 312 because the condition on line 311 was never true

312 return 

313 # ir_actions is inherited, so foreign key doesn't work on it 

314 if not comodel._auto or comodel._table == 'ir_actions': 

315 return 

316 # create/update the foreign key, and reflect it in 'ir.model.constraint' 

317 model.pool.add_foreign_key( 

318 model._table, self.name, comodel._table, 'id', self.ondelete or 'set null', 

319 model, self._module 

320 ) 

321 

322 def _update_inverse(self, records, value): 

323 for record in records: 

324 self._update_cache(record, self.convert_to_cache(value, record, validate=False)) 

325 

326 def convert_to_column(self, value, record, values=None, validate=True): 

327 return value or None 

328 

329 def convert_to_cache(self, value, record, validate=True): 

330 # cache format: id or None 

331 if type(value) is int or type(value) is NewId: 

332 id_ = value 

333 elif isinstance(value, BaseModel): 

334 if validate and (value._name != self.comodel_name or len(value) > 1): 334 ↛ 335line 334 didn't jump to line 335 because the condition on line 334 was never true

335 raise ValueError("Wrong value for %s: %r" % (self, value)) 

336 id_ = value._ids[0] if value._ids else None 

337 elif isinstance(value, tuple): 337 ↛ 339line 337 didn't jump to line 339 because the condition on line 337 was never true

338 # value is either a pair (id, name), or a tuple of ids 

339 id_ = value[0] if value else None 

340 elif isinstance(value, dict): 340 ↛ 342line 340 didn't jump to line 342 because the condition on line 340 was never true

341 # return a new record (with the given field 'id' as origin) 

342 comodel = record.env[self.comodel_name] 

343 origin = comodel.browse(value.get('id')) 

344 id_ = comodel.new(value, origin=origin).id 

345 else: 

346 id_ = None 

347 

348 if self.delegate and record and not any(record._ids): 

349 # if all records are new, then so is the parent 

350 id_ = id_ and NewId(id_) 

351 

352 return id_ 

353 

354 def convert_to_record(self, value, record): 

355 # use registry to avoid creating a recordset for the model 

356 ids = () if value is None else (value,) 

357 prefetch_ids = PrefetchMany2one(record, self) 

358 return record.pool[self.comodel_name](record.env, ids, prefetch_ids) 

359 

360 def convert_to_record_multi(self, values, records): 

361 # return the ids as a recordset without duplicates 

362 prefetch_ids = PrefetchMany2one(records, self) 

363 ids = tuple(unique(id_ for id_ in values if id_ is not None)) 

364 return records.pool[self.comodel_name](records.env, ids, prefetch_ids) 

365 

366 def convert_to_read(self, value, record, use_display_name=True): 

367 if use_display_name and value: 

368 # evaluate display_name as superuser, because the visibility of a 

369 # many2one field value (id and name) depends on the current record's 

370 # access rights, and not the value's access rights. 

371 try: 

372 # performance: value.sudo() prefetches the same records as value 

373 return (value.id, value.sudo().display_name) 

374 except MissingError: 

375 # Should not happen, unless the foreign key is missing. 

376 return False 

377 else: 

378 return value.id 

379 

380 def convert_to_write(self, value, record): 

381 if type(value) is int or type(value) is NewId: 

382 return value 

383 if not value: 

384 return False 

385 if isinstance(value, BaseModel) and value._name == self.comodel_name: 385 ↛ 387line 385 didn't jump to line 387 because the condition on line 385 was always true

386 return value.id 

387 if isinstance(value, tuple): 

388 # value is either a pair (id, name), or a tuple of ids 

389 return value[0] if value else False 

390 if isinstance(value, dict): 

391 return record.env[self.comodel_name].new(value).id 

392 raise ValueError("Wrong value for %s: %r" % (self, value)) 

393 

394 def convert_to_export(self, value, record): 

395 return value.display_name if value else '' 

396 

397 def convert_to_display_name(self, value, record): 

398 return value.display_name 

399 

400 def write(self, records, value): 

401 # discard recomputation of self on records 

402 records.env.remove_to_compute(self, records) 

403 

404 # discard the records that are not modified 

405 cache_value = self.convert_to_cache(value, records) 

406 records = self._filter_not_equal(records, cache_value) 

407 if not records: 

408 return 

409 

410 # remove records from the cache of one2many fields of old corecords 

411 self._remove_inverses(records, cache_value) 

412 

413 # update the cache of self 

414 self._update_cache(records, cache_value, dirty=True) 

415 

416 # update the cache of one2many fields of new corecord 

417 self._update_inverses(records, cache_value) 

418 

419 def _remove_inverses(self, records: BaseModel, value): 

420 """ Remove `records` from the cached values of the inverse fields (o2m) of `self`. """ 

421 inverse_fields = records.pool.field_inverses[self] 

422 if not inverse_fields: 

423 return 

424 

425 record_ids = set(records._ids) 

426 # align(id) returns a NewId if records are new, a real id otherwise 

427 align = (lambda id_: id_) if all(record_ids) else (lambda id_: id_ and NewId(id_)) 

428 field_cache = self._get_cache(records.env) 

429 corecords = records.env[self.comodel_name].browse( 

430 align(coid) for record_id in records._ids 

431 if (coid := field_cache.get(record_id)) is not None 

432 ) 

433 

434 for invf in inverse_fields: 

435 inv_cache = invf._get_cache(corecords.env) 

436 for corecord in corecords: 

437 ids0 = inv_cache.get(corecord.id) 

438 if ids0 is not None: 

439 ids1 = tuple(id_ for id_ in ids0 if id_ not in record_ids) 

440 invf._update_cache(corecord, ids1) 

441 

442 def _update_inverses(self, records: BaseModel, value): 

443 """ Add `records` to the cached values of the inverse fields (o2m) of `self`. """ 

444 if value is None: 

445 return 

446 corecord = self.convert_to_record(value, records) 

447 for invf in records.pool.field_inverses[self]: 

448 valid_records = records.filtered_domain(invf.get_comodel_domain(corecord)) 

449 if not valid_records: 

450 continue 

451 ids0 = invf._get_cache(corecord.env).get(corecord.id) 

452 # if the value for the corecord is not in cache, but this is a new 

453 # record, assign it anyway, as you won't be able to fetch it from 

454 # database (see `test_sale_order`) 

455 if ids0 is not None or not corecord.id: 

456 ids1 = tuple(unique((ids0 or ()) + valid_records._ids)) 

457 invf._update_cache(corecord, ids1) 

458 

459 def to_sql(self, model: BaseModel, alias: str) -> SQL: 

460 sql_field = super().to_sql(model, alias) 

461 if self.company_dependent: 

462 comodel = model.env[self.comodel_name] 

463 sql_field = SQL( 

464 '''(SELECT %(cotable_alias)s.id 

465 FROM %(cotable)s AS %(cotable_alias)s 

466 WHERE %(cotable_alias)s.id = %(ref)s)''', 

467 cotable=SQL.identifier(comodel._table), 

468 cotable_alias=SQL.identifier(Query.make_alias(comodel._table, 'exists')), 

469 ref=sql_field, 

470 ) 

471 return sql_field 

472 

473 def condition_to_sql(self, field_expr: str, operator: str, value, model: BaseModel, alias: str, query: Query) -> SQL: 

474 if operator not in ('any', 'not any', 'any!', 'not any!') or field_expr != self.name: 

475 # for other operators than 'any', just generate condition based on column type 

476 return super().condition_to_sql(field_expr, operator, value, model, alias, query) 

477 

478 comodel = model.env[self.comodel_name] 

479 sql_field = model._field_to_sql(alias, field_expr, query) 

480 can_be_null = self not in model.env.registry.not_null_fields 

481 bypass_access = operator in ('any!', 'not any!') or self.bypass_search_access 

482 positive = operator in ('any', 'any!') 

483 

484 # Decide whether to use a LEFT JOIN 

485 left_join = bypass_access and isinstance(value, Domain) 

486 if left_join and not positive: 

487 # For 'not any!', we get a better query with a NOT IN when we have a 

488 # lot of positive conditions which have a better chance to use 

489 # indexes. 

490 # `field NOT IN (SELECT ... WHERE z = y)` better than 

491 # `LEFT JOIN ... ON field = id WHERE z <> y` 

492 # There are some exceptions: we filter on 'id'. 

493 left_join = sum( 

494 (-1 if cond.operator in Domain.NEGATIVE_OPERATORS else 1) 

495 for cond in value.iter_conditions() 

496 ) < 0 or any( 

497 cond.field_expr == 'id' and cond.operator not in Domain.NEGATIVE_OPERATORS 

498 for cond in value.iter_conditions() 

499 ) 

500 

501 if left_join: 

502 comodel, coalias = self.join(model, alias, query) 

503 if not positive: 503 ↛ 504line 503 didn't jump to line 504 because the condition on line 503 was never true

504 value = (~value).optimize_full(comodel) 

505 sql = value._to_sql(comodel, coalias, query) 

506 if self.company_dependent: 506 ↛ 507line 506 didn't jump to line 507 because the condition on line 506 was never true

507 sql = self._condition_to_sql_company(sql, field_expr, operator, value, model, alias, query) 

508 if can_be_null: 

509 if positive: 509 ↛ 512line 509 didn't jump to line 512 because the condition on line 509 was always true

510 sql = SQL("(%s IS NOT NULL AND %s)", sql_field, sql) 

511 else: 

512 sql = SQL("(%s IS NULL OR %s)", sql_field, sql) 

513 return sql 

514 

515 if isinstance(value, Domain): 515 ↛ 517line 515 didn't jump to line 517 because the condition on line 515 was always true

516 value = comodel._search(value, active_test=False, bypass_access=bypass_access) 

517 if isinstance(value, Query): 517 ↛ 519line 517 didn't jump to line 519 because the condition on line 517 was always true

518 subselect = value.subselect() 

519 elif isinstance(value, SQL): 

520 subselect = SQL("(%s)", value) 

521 else: 

522 raise TypeError(f"condition_to_sql() 'any' operator accepts Domain, SQL or Query, got {value}") 

523 sql = SQL( 

524 "%s%s%s", 

525 sql_field, 

526 SQL(" IN ") if positive else SQL(" NOT IN "), 

527 subselect, 

528 ) 

529 if can_be_null and not positive: 

530 sql = SQL("(%s IS NULL OR %s)", sql_field, sql) 

531 if self.company_dependent: 531 ↛ 532line 531 didn't jump to line 532 because the condition on line 531 was never true

532 sql = self._condition_to_sql_company(sql, field_expr, operator, value, model, alias, query) 

533 return sql 

534 

535 def join(self, model: BaseModel, alias: str, query: Query) -> tuple[BaseModel, str]: 

536 """ Add a LEFT JOIN to ``query`` by following field ``self``, 

537 and return the joined table's corresponding model and alias. 

538 """ 

539 comodel = model.env[self.comodel_name] 

540 coalias = query.make_alias(alias, self.name) 

541 query.add_join('LEFT JOIN', coalias, comodel._table, SQL( 

542 "%s = %s", 

543 model._field_to_sql(alias, self.name, query), 

544 SQL.identifier(coalias, 'id'), 

545 )) 

546 return (comodel, coalias) 

547 

548 

549class _RelationalMulti(_Relational): 

550 r"Abstract class for relational fields \*2many." 

551 write_sequence = 20 

552 

553 # Important: the cache contains the ids of all the records in the relation, 

554 # including inactive records. Inactive records are filtered out by 

555 # convert_to_record(), depending on the context. 

556 

557 def _update_inverse(self, records, value): 

558 new_id = value.id 

559 assert not new_id, "Field._update_inverse can only be called with a new id" 

560 field_cache = self._get_cache(records.env) 

561 for record_id in records._ids: 

562 assert not record_id, "Field._update_inverse can only be called with new records" 

563 cache_value = field_cache.get(record_id, SENTINEL) 

564 if cache_value is SENTINEL: 564 ↛ 567line 564 didn't jump to line 567 because the condition on line 564 was always true

565 records.env.transaction.field_data_patches[self][record_id].append(new_id) 

566 else: 

567 field_cache[record_id] = tuple(unique(cache_value + (new_id,))) 

568 

569 def _update_cache(self, records, cache_value, dirty=False): 

570 field_patches = records.env.transaction.field_data_patches.get(self) 

571 if field_patches and records: 

572 for record in records: 

573 ids = field_patches.pop(record.id, ()) 

574 if ids: 574 ↛ 575line 574 didn't jump to line 575 because the condition on line 574 was never true

575 value = tuple(unique(itertools.chain(cache_value, ids))) 

576 else: 

577 value = cache_value 

578 super()._update_cache(record, value, dirty) 

579 return 

580 super()._update_cache(records, cache_value, dirty) 

581 

582 def convert_to_cache(self, value, record, validate=True): 

583 # cache format: tuple(ids) 

584 if isinstance(value, BaseModel): 

585 if validate and value._name != self.comodel_name: 585 ↛ 586line 585 didn't jump to line 586 because the condition on line 585 was never true

586 raise ValueError("Wrong value for %s: %s" % (self, value)) 

587 ids = value._ids 

588 if record and not record.id: 588 ↛ 590line 588 didn't jump to line 590 because the condition on line 588 was never true

589 # x2many field value of new record is new records 

590 ids = tuple(it and NewId(it) for it in ids) 

591 return ids 

592 

593 elif isinstance(value, (list, tuple)): 

594 # value is a list/tuple of commands, dicts or record ids 

595 comodel = record.env[self.comodel_name] 

596 # if record is new, the field's value is new records 

597 if record and not record.id: 

598 browse = lambda it: comodel.browse((it and NewId(it),)) 

599 else: 

600 browse = comodel.browse 

601 # determine the value ids: in case of a real record or a new record 

602 # with origin, take its current value 

603 ids = OrderedSet(record[self.name]._ids if record._origin else ()) 

604 # modify ids with the commands 

605 for command in value: 

606 if isinstance(command, (tuple, list)): 

607 if command[0] == Command.CREATE: 

608 ids.add(comodel.new(command[2], ref=command[1]).id) 

609 elif command[0] == Command.UPDATE: 609 ↛ 610line 609 didn't jump to line 610 because the condition on line 609 was never true

610 line = browse(command[1]) 

611 if validate: 

612 line.update(command[2]) 

613 else: 

614 line._update_cache(command[2], validate=False) 

615 ids.add(line.id) 

616 elif command[0] in (Command.DELETE, Command.UNLINK): 616 ↛ 617line 616 didn't jump to line 617 because the condition on line 616 was never true

617 ids.discard(browse(command[1]).id) 

618 elif command[0] == Command.LINK: 

619 ids.add(browse(command[1]).id) 

620 elif command[0] == Command.CLEAR: 620 ↛ 621line 620 didn't jump to line 621 because the condition on line 620 was never true

621 ids.clear() 

622 elif command[0] == Command.SET: 622 ↛ 605line 622 didn't jump to line 605 because the condition on line 622 was always true

623 ids = OrderedSet(browse(it).id for it in command[2]) 

624 elif isinstance(command, dict): 624 ↛ 625line 624 didn't jump to line 625 because the condition on line 624 was never true

625 ids.add(comodel.new(command).id) 

626 else: 

627 ids.add(browse(command).id) 

628 # return result as a tuple 

629 return tuple(ids) 

630 

631 elif not value: 631 ↛ 634line 631 didn't jump to line 634 because the condition on line 631 was always true

632 return () 

633 

634 raise ValueError("Wrong value for %s: %s" % (self, value)) 

635 

636 def convert_to_record(self, value, record): 

637 # use registry to avoid creating a recordset for the model 

638 prefetch_ids = PrefetchX2many(record, self) 

639 Comodel = record.pool[self.comodel_name] 

640 corecords = Comodel(record.env, value, prefetch_ids) 

641 if ( 

642 Comodel._active_name 

643 and self.context.get('active_test', record.env.context.get('active_test', True)) 

644 ): 

645 corecords = corecords.filtered(Comodel._active_name).with_prefetch(prefetch_ids) 

646 return corecords 

647 

648 def convert_to_record_multi(self, values, records): 

649 # return the list of ids as a recordset without duplicates 

650 prefetch_ids = PrefetchX2many(records, self) 

651 Comodel = records.pool[self.comodel_name] 

652 ids = tuple(unique(id_ for ids in values for id_ in ids)) 

653 corecords = Comodel(records.env, ids, prefetch_ids) 

654 if ( 

655 Comodel._active_name 

656 and self.context.get('active_test', records.env.context.get('active_test', True)) 

657 ): 

658 corecords = corecords.filtered(Comodel._active_name).with_prefetch(prefetch_ids) 

659 return corecords 

660 

661 def convert_to_read(self, value, record, use_display_name=True): 

662 return value.ids 

663 

664 def convert_to_write(self, value, record): 

665 if isinstance(value, tuple): 

666 # a tuple of ids, this is the cache format 

667 value = record.env[self.comodel_name].browse(value) 

668 

669 if isinstance(value, BaseModel) and value._name == self.comodel_name: 

670 def get_origin(val): 

671 return val._origin if isinstance(val, BaseModel) else val 

672 

673 # make result with new and existing records 

674 inv_names = {field.name for field in record.pool.field_inverses[self]} 

675 result = [Command.set([])] 

676 for record in value: 

677 origin = record._origin 

678 if not origin: 678 ↛ 679line 678 didn't jump to line 679 because the condition on line 678 was never true

679 values = record._convert_to_write({ 

680 name: record[name] 

681 for name in record._cache 

682 if name not in inv_names 

683 }) 

684 result.append(Command.create(values)) 

685 else: 

686 result[0][2].append(origin.id) 

687 if record != origin: 

688 values = record._convert_to_write({ 

689 name: record[name] 

690 for name in record._cache 

691 if name not in inv_names and get_origin(record[name]) != origin[name] 

692 }) 

693 if values: 693 ↛ 694line 693 didn't jump to line 694 because the condition on line 693 was never true

694 result.append(Command.update(origin.id, values)) 

695 return result 

696 

697 if value is False or value is None: 697 ↛ 698line 697 didn't jump to line 698 because the condition on line 697 was never true

698 return [Command.clear()] 

699 

700 if isinstance(value, list): 700 ↛ 703line 700 didn't jump to line 703 because the condition on line 700 was always true

701 return value 

702 

703 raise ValueError("Wrong value for %s: %s" % (self, value)) 

704 

705 def convert_to_export(self, value, record): 

706 return ','.join(value.mapped('display_name')) if value else '' 

707 

708 def convert_to_display_name(self, value, record):