# This file is part of Tryton. The COPYRIGHT file at the top level of # this repository contains the full copyright notices and license terms. ''' A library to access Tryton's models like a client. ''' import datetime import functools import threading from decimal import Decimal from functools import total_ordering import proteus.config __version__ = "7.9.0" __all__ = ['Model', 'Wizard', 'Report'] _MODELS = threading.local() class _EvalEnvironment(dict): 'Dictionary for evaluation' __slots__ = ('parent', 'eval_type') def __init__(self, parent, eval_type='eval'): super().__init__() self.parent = parent assert eval_type in ('eval', 'on_change') self.eval_type = eval_type def __getitem__(self, item): if item == 'id': return self.parent.id if item == '_parent_' + self.parent._parent_name \ and self.parent._parent: return _EvalEnvironment(self.parent._parent, eval_type=self.eval_type) if self.eval_type == 'eval': return self.parent._get_eval()[item] else: return self.parent._get_on_change_values(fields=[item])[item] def __getattr__(self, item): try: return self.__getitem__(item) except KeyError: raise AttributeError(item) def get(self, item, default=None): try: return self.__getitem__(item) except KeyError: pass return super().get(item, default) def __bool__(self): return True def __str__(self): return str(self.parent) def __repr__(self): return repr(self.parent) def __contains__(self, item): if item == 'id': return True if item == '_parent_' + self.parent._parent_name \ and self.parent._parent: return True if self.eval_type == 'eval': return item in self.parent._get_eval() else: return item in self.parent._fields class dualmethod(object): """Descriptor implementing combination of class and instance method When called on an instance, the class is passed as the first argument and a list with the instance as the second. When called on a class, the class itself is passed as the first argument. >>> class Example(object): ... @dualmethod ... def method(cls, instances): ... print(len(instances)) ... >>> Example.method([Example()]) 1 >>> Example().method() 1 """ __slots__ = ('func') def __init__(self, func): self.func = func def __get__(self, instance, owner): @functools.wraps(self.func) def newfunc(*args, **kwargs): if instance: return self.func(owner, [instance], *args, **kwargs) else: return self.func(owner, *args, **kwargs) return newfunc class FieldDescriptor(object): default = None def __init__(self, name, definition): super().__init__() self.name = name self.definition = definition self.__doc__ = definition['string'] def __get__(self, instance, owner): if instance.id >= 0: instance._read(self.name) return instance._values.get(self.name, self.default) def __set__(self, instance, value): if instance.id >= 0: instance._read(self.name) previous = getattr(instance, self.name) instance._values[self.name] = value if previous != getattr(instance, self.name): instance._changed.add(self.name) instance._on_change([self.name]) if instance._parent: instance._parent._changed.add(instance._parent_field_name) instance._parent._on_change([instance._parent_field_name]) class BooleanDescriptor(FieldDescriptor): default = False def __set__(self, instance, value): assert isinstance(value, bool) super().__set__(instance, value) class CharDescriptor(FieldDescriptor): default = None def __set__(self, instance, value): assert isinstance(value, str) or value is None if self.definition.get('strip') and value: if self.definition['strip'] == 'leading': value = value.lstrip() elif self.definition['strip'] == 'trailing': value = value.rstrip() else: value = value.strip() super().__set__(instance, value or '') class BinaryDescriptor(FieldDescriptor): default = None def __set__(self, instance, value): assert isinstance(value, (bytes, bytearray)) or value is None super().__set__(instance, value) class IntegerDescriptor(FieldDescriptor): def __set__(self, instance, value): assert isinstance(value, (int, type(None))) super().__set__(instance, value) class FloatDescriptor(FieldDescriptor): def __set__(self, instance, value): assert isinstance(value, (int, float, Decimal, type(None))) if value is not None: value = float(value) super().__set__(instance, value) class NumericDescriptor(FieldDescriptor): def __set__(self, instance, value): assert isinstance(value, (type(None), Decimal)) # TODO add digits validation super().__set__(instance, value) class SelectionDescriptor(FieldDescriptor): def __set__(self, instance, value): assert isinstance(value, str) or value is None # TODO add selection validation super().__set__(instance, value) class MultiSelectionDescriptor(FieldDescriptor): def __set__(self, instance, value): assert isinstance(value, (list, type(None))) # TODO add selection validation super().__set__(instance, value) class ReferenceDescriptor(FieldDescriptor): def __get__(self, instance, owner): value = super().__get__(instance, owner) if isinstance(value, str): model_name, id = value.split(',', 1) if model_name: Relation = Model.get(model_name, instance._config) config = Relation._config with config.reset_context(), \ config.set_context(instance._context): value = Relation(int(id)) instance._values[self.name] = value return value def __set__(self, instance, value): assert isinstance(value, (Model, type(None), str)) if isinstance(value, str): assert value.startswith(',') elif isinstance(value, Model): assert value._config == instance._config super().__set__(instance, value) class DateDescriptor(FieldDescriptor): def __get__(self, instance, owner): value = super().__get__(instance, owner) if isinstance(value, datetime.datetime): value = value.date() instance._values[self.name] = value return value def __set__(self, instance, value): assert isinstance(value, datetime.date) or value is None super().__set__(instance, value) class DateTimeDescriptor(FieldDescriptor): def __set__(self, instance, value): assert isinstance(value, datetime.datetime) or value is None super().__set__(instance, value) class TimeDescriptor(FieldDescriptor): def __set__(self, instance, value): assert isinstance(value, datetime.time) or value is None super().__set__(instance, value) class TimeDeltaDescriptor(FieldDescriptor): def __set__(self, instance, value): assert isinstance(value, datetime.timedelta) or value is None super().__set__(instance, value) class DictDescriptor(FieldDescriptor): def __get__(self, instance, owner): value = super().__get__(instance, owner) if value: value = value.copy() return value def __set__(self, instance, value): assert isinstance(value, dict) or value is None super().__set__(instance, value) class Many2OneDescriptor(FieldDescriptor): def __get__(self, instance, owner): Relation = Model.get(self.definition['relation'], instance._config) value = super().__get__(instance, owner) if isinstance(value, int): config = Relation._config with config.reset_context(), config.set_context(instance._context): value = Relation(value) if self.name in instance._values: instance._values[self.name] = value return value def __set__(self, instance, value): assert isinstance(value, (Model, type(None))) if value: assert value._config == instance._config super().__set__(instance, value) class One2OneDescriptor(Many2OneDescriptor): pass class One2ManyDescriptor(FieldDescriptor): default = [] def __get__(self, instance, owner): from .pyson import PYSONDecoder Relation = Model.get(self.definition['relation'], instance._config) value = super().__get__(instance, owner) if not isinstance(value, ModelList): ctx = instance._context.copy() if instance._context else {} if self.definition.get('context'): decoder = PYSONDecoder(_EvalEnvironment(instance)) ctx.update(decoder.decode(self.definition.get('context'))) config = Relation._config with config.reset_context(), config.set_context(ctx): value = ModelList(self.definition, (Relation(id) for id in value or []), instance, self.name) instance._values[self.name] = value return value def __set__(self, instance, value): raise AttributeError class Many2ManyDescriptor(One2ManyDescriptor): pass class ValueDescriptor(object): def __init__(self, name, definition): super().__init__() self.name = name self.definition = definition def __get__(self, instance, owner): return getattr(instance, self.name) class ReferenceValueDescriptor(ValueDescriptor): def __get__(self, instance, owner): value = super().__get__(instance, owner) if isinstance(value, Model): value = '%s,%s' % (value.__class__.__name__, value.id) return value or None class Many2OneValueDescriptor(ValueDescriptor): def __get__(self, instance, owner): value = super().__get__(instance, owner) return value and value.id or None class One2OneValueDescriptor(Many2OneValueDescriptor): pass class One2ManyValueDescriptor(ValueDescriptor): def __get__(self, instance, owner): value = [] value_list = getattr(instance, self.name) parent_name = self.definition.get('relation_field', '') to_add = [] to_create = [] to_write = [] for record in value_list: if record.id >= 0: values = record._get_values(fields=record._changed) values.pop(parent_name, None) if record._changed and values: to_write.extend(([record.id], values)) to_add.append(record.id) else: values = record._get_values() values.pop(parent_name, None) to_create.append(values) if to_add: value.append(('add', to_add)) if to_create: value.append(('create', to_create)) if to_write: value.append(('write',) + tuple(to_write)) if value_list.record_removed: value.append(('remove', [x.id for x in value_list.record_removed])) if value_list.record_deleted: value.append(('delete', [x.id for x in value_list.record_deleted])) return value class Many2ManyValueDescriptor(One2ManyValueDescriptor): pass class EvalDescriptor(object): def __init__(self, name, definition): super().__init__() self.name = name self.definition = definition def __get__(self, instance, owner): return getattr(instance, self.name) class ReferenceEvalDescriptor(EvalDescriptor): def __get__(self, instance, owner): value = super().__get__(instance, owner) if isinstance(value, Model): value = '%s,%s' % (value.__class__.__name__, value.id) return value or None class Many2OneEvalDescriptor(EvalDescriptor): def __get__(self, instance, owner): value = super().__get__(instance, owner) if value: return value.id return None class One2OneEvalDescriptor(Many2OneEvalDescriptor): pass class One2ManyEvalDescriptor(EvalDescriptor): def __get__(self, instance, owner): # Directly use _values to prevent infinite recursion with # One2ManyDescriptor which could evaluate this field to decode the # context value = instance._values.get(self.name, []) if isinstance(value, ModelList): return [x.id for x in value] else: return value class Many2ManyEvalDescriptor(One2ManyEvalDescriptor): pass class MetaModelFactory(object): descriptors = { 'boolean': BooleanDescriptor, 'char': CharDescriptor, 'text': CharDescriptor, 'binary': BinaryDescriptor, 'selection': SelectionDescriptor, 'multiselection': MultiSelectionDescriptor, 'integer': IntegerDescriptor, 'biginteger': IntegerDescriptor, 'float': FloatDescriptor, 'numeric': NumericDescriptor, 'reference': ReferenceDescriptor, 'date': DateDescriptor, 'datetime': DateTimeDescriptor, 'timestamp': DateTimeDescriptor, 'time': TimeDescriptor, 'timedelta': TimeDeltaDescriptor, 'dict': DictDescriptor, 'many2one': Many2OneDescriptor, 'one2many': One2ManyDescriptor, 'many2many': Many2ManyDescriptor, 'one2one': One2OneDescriptor, } value_descriptors = { 'reference': ReferenceValueDescriptor, 'many2one': Many2OneValueDescriptor, 'one2many': One2ManyValueDescriptor, 'many2many': Many2ManyValueDescriptor, 'one2one': One2OneValueDescriptor, } eval_descriptors = { 'reference': ReferenceEvalDescriptor, 'many2one': Many2OneEvalDescriptor, 'one2many': One2ManyEvalDescriptor, 'many2many': Many2ManyEvalDescriptor, 'one2one': One2OneEvalDescriptor, } def __init__(self, model_name, config=None): super().__init__() self.model_name = model_name self.config = config or proteus.config.get_config() def __call__(self): models_key = 'c%su%s' % (id(self.config), self.config.user) if not hasattr(_MODELS, models_key): setattr(_MODELS, models_key, {}) class MetaModel(type): 'Meta class for Model' __slots__ = () def __new__(mcs, name, bases, dict): if self.model_name in getattr(_MODELS, models_key): return getattr(_MODELS, models_key)[self.model_name] proxy = self.config.get_proxy(self.model_name) context = self.config.context name = self.model_name dict['_proxy'] = proxy dict['_config'] = self.config dict['_fields'] = proxy.fields_get(None, context) for field_name, definition in dict['_fields'].items(): if field_name == 'id': continue Descriptor = self.descriptors.get(definition['type'], FieldDescriptor) dict[field_name] = Descriptor(field_name, definition) VDescriptor = self.value_descriptors.get( definition['type'], ValueDescriptor) dict['__%s_value' % field_name] = VDescriptor( field_name, definition) EDescriptor = self.eval_descriptors.get( definition['type'], EvalDescriptor) dict['__%s_eval' % field_name] = EDescriptor( field_name, definition) for method in self.config.get_proxy_methods(self.model_name): setattr(mcs, method, getattr(proxy, method)) res = type.__new__(mcs, name, bases, dict) getattr(_MODELS, models_key)[self.model_name] = res return res __new__.__doc__ = type.__new__.__doc__ return MetaModel class ModelList(list): 'List for Model' __slots__ = ('model_name', 'parent', 'parent_field_name', 'parent_name', 'domain', 'context', 'add_remove', 'search_order', 'search_context', 'record_removed', 'record_deleted') def __init__(self, definition, sequence=None, parent=None, parent_field_name=''): self.model_name = definition['relation'] if sequence is None: sequence = [] self.parent = parent if parent: assert parent_field_name self.parent_field_name = parent_field_name self.parent_name = definition.get('relation_field', '') self.domain = definition.get('domain', []) self.context = definition.get('context') self.add_remove = definition.get('add_remove') self.search_order = definition.get('search_order', 'null') self.search_context = definition.get('search_context', '{}') self.record_removed = set() self.record_deleted = set() result = super().__init__(sequence) assert len(self) == len(set(self)) self.__check(self, on_change=False) return result __init__.__doc__ = list.__init__.__doc__ def _changed(self): 'Signal change to parent' if self.parent: self.parent._changed.add(self.parent_field_name) self.parent._on_change([self.parent_field_name]) def _get_context(self): from .pyson import PYSONDecoder decoder = PYSONDecoder(_EvalEnvironment(self.parent)) ctx = self.parent._context.copy() if self.parent._context else {} ctx.update(decoder.decode(self.context) if self.context else {}) return ctx def __check(self, records, on_change=True): config = None for record in records: assert isinstance(record, Model) assert record.__class__.__name__ == self.model_name if self.parent: assert record._config == self.parent._config elif self: assert record._config == self[0]._config elif config: assert record._config == config else: config = record._config if record._group is not self: assert record._group is None record._group = self for record in records: # Set parent field to trigger on_change if (on_change and self.parent and self.parent_name in record._fields): definition = record._fields[self.parent_name] if definition['type'] in ('many2one', 'reference'): setattr(record, self.parent_name, self.parent) self.record_removed.difference_update(records) self.record_deleted.difference_update(records) def append(self, record): self.__check([record]) if record not in self: super().append(record) self._changed() append.__doc__ = list.append.__doc__ def extend(self, iterable): iterable = list(iterable) self.__check(iterable) set_ = set(self) super().extend(r for r in iterable if r not in set_) self._changed() extend.__doc__ = list.extend.__doc__ def insert(self, index, record): raise NotImplementedError insert.__doc__ = list.insert.__doc__ def pop(self, index=-1, _changed=True): self.record_removed.add(self[index]) self[index]._group = None res = super().pop(index) if _changed: self._changed() return res pop.__doc__ = list.pop.__doc__ def remove(self, record, _changed=True): if record.id >= 0: self.record_deleted.add(record) record._group = None res = super().remove(record) if _changed: self._changed() return res remove.__doc__ = list.remove.__doc__ def reverse(self): raise NotImplementedError reverse.__doc__ = list.reverse.__doc__ def sort(self): raise NotImplementedError sort.__doc__ = list.sort.__doc__ def new(self, **kwargs): 'Adds a new record to the ModelList and returns it' Relation = Model.get(self.model_name, self.parent._config) config = Relation._config with config.reset_context(), config.set_context(self._get_context()): # Set parent for on_change calls from default_get new_record = Relation(_group=self, **kwargs) self.append(new_record) return new_record def find(self, condition=None, offset=0, limit=None, order=None): 'Returns records matching condition taking into account list domain' from .pyson import PYSONDecoder decoder = PYSONDecoder(_EvalEnvironment(self.parent)) Relation = Model.get(self.model_name, self.parent._config) if condition is None: condition = [] field_domain = decoder.decode(self.domain) add_remove_domain = (decoder.decode(self.add_remove) if self.add_remove else []) new_domain = [field_domain, add_remove_domain, condition] context = self._get_context() context.update(decoder.decode(self.search_context)) order = order if order else decoder.decode(self.search_order) config = Relation._config with config.reset_context(), config.set_context(context): return Relation.find(new_domain, offset, limit, order) def set_sequence(self, field='sequence'): changed = False prev = None for record in self: if prev: index = getattr(prev, field) else: index = None update = False value = getattr(record, field) if value is None: if index: update = True elif prev and record.id >= 0: update = record.id < prev.id elif value == index: if prev and record.id >= 0: update = record.id < prev.id elif value <= (index or 0): update = True if update: if index is None: index = 0 index += 1 setattr(record, field, index) changed = record prev = record if changed: self._changed() @total_ordering class Model(object): 'Model class for Tryton records' __slots__ = ('__id', '_values', '_changed', '_group', '__context') __counter = -1 _proxy = None _config = None _fields = None def __init__(self, id=None, _default=True, _group=None, **kwargs): super().__init__() if id is not None: assert not kwargs self.__id = id if id is not None else Model.__counter if self.__id < 0: Model.__counter -= 1 self._values = {} # store the values of fields self._changed = set() # store the changed fields self._group = _group # store the parent group self.__context = self._config.context # store the context if self.id < 0 and _default: self._default_get() for field_name, value in kwargs.items(): definition = self._fields[field_name] if definition['type'] in ('one2many', 'many2many'): relation = Model.get(definition['relation'], self._config) def instantiate(v): if isinstance(v, int): return relation(v) elif isinstance(v, dict): return relation(_default=_default, **v) else: return v value = [instantiate(x) for x in value] getattr(self, field_name).extend(value) else: if definition['type'] == 'many2one': if isinstance(value, int): relation = Model.get( definition['relation'], self._config) value = relation(value) setattr(self, field_name, value) __init__.__doc__ = object.__init__.__doc__ @property def _parent(self): if self._group is not None: return self._group.parent @property def _parent_field_name(self): if self._group is not None: return self._group.parent_field_name return '' @property def _parent_name(self): if self._group is not None: return self._group.parent_name return '' @property def _context(self): if self._group: context = self._group._get_context() else: context = self.__context return context @classmethod def get(cls, name, config=None): 'Get a class for the named Model' if (bytes == str) and isinstance(name, str): name = name.encode('utf-8') class Spam(Model, metaclass=MetaModelFactory(name, config=config)()): __slots__ = () return Spam @classmethod def reset(cls, config=None, *names): 'Reset class definition for Models named' config = config or proteus.config.get_config() models_key = 'c%su%s' % (id(config), config.user) if not names: setattr(_MODELS, models_key, {}) else: models = getattr(_MODELS, models_key, {}) for name in names: del models[name] def __str__(self): return '%s,%d' % (self.__class__.__name__, self.id) def __repr__(self): if self._config == proteus.config.get_config(): return "proteus.Model.get('%s')(%d)" % (self.__class__.__name__, self.id) return "proteus.Model.get('%s', %s)(%d)" % (self.__class__.__name__, repr(self._config), self.id) def __eq__(self, other): if isinstance(other, Model): return ((self.__class__.__name__, self.id) == (other.__class__.__name__, other.id)) return NotImplemented def __lt__(self, other): if not isinstance(other, Model) or self.__class__ != other.__class__: return NotImplemented return self.id < other.id def __hash__(self): return hash(self.__class__.__name__) ^ hash(self.id) def __int__(self): return self.id @property def id(self): 'The unique ID' return self.__id @id.setter def id(self, value): assert self.__id < 0 self.__id = int(value) @classmethod def find(cls, condition=None, offset=0, limit=None, order=None): 'Return records matching condition' if condition is None: condition = [] ids = cls._proxy.search(condition, offset, limit, order, cls._config.context) return [cls(id) for id in ids] @dualmethod def reload(cls, records): 'Reload record' for record in records: record._values = {} record._changed = set() @dualmethod def save(cls, records): 'Save records' if not records: return proxy = records[0]._proxy config = records[0]._config context = records[0]._context.copy() create, write = [], [] for record in records: assert proxy == record._proxy assert config == record._config assert context == record._context if record.id < 0: create.append(record) elif record._changed: write.append(record) if create: values = [r._get_values() for r in create] ids = proxy.create(values, context) for record, id_ in zip(create, ids): record.id = id_ if write: values = [] context['_timestamp'] = {} for record in write: values.append([record.id]) values.append(record._get_values(fields=record._changed)) context['_timestamp'].update(record._get_timestamp()) values.append(context) proxy.write(*values) for record in records: record.reload() @dualmethod def delete(cls, records): 'Delete records' if not records: return proxy = records[0]._proxy config = records[0]._config context = records[0]._context.copy() timestamp = {} delete = [] for record in records: assert proxy == record._proxy assert config == record._config assert context == record._context if record.id >= 0: timestamp.update(record._get_timestamp()) delete.append(record.id) context['_timestamp'] = timestamp if delete: proxy.delete(delete, context) cls.reload(records) @dualmethod def duplicate(cls, records, default=None): 'Duplicate the record' ids = cls._proxy.copy([r.id for r in records], default, cls._config.context) return [cls(id) for id in ids] @dualmethod def click(cls, records, button, change=None): 'Click on button' if not records: return proxy = records[0]._proxy config = records[0]._config context = records[0]._context.copy() for record in records: assert proxy == record._proxy assert config == record._config assert context == record._context if change is None: cls.save(records) cls.reload(records) # Force reload because save doesn't always record_ids = [r.id for r in records] action = getattr(proxy, button)(record_ids, context) if action and not isinstance(action, str): if isinstance(action, int): action = config.get_proxy('ir.action').get_action_value( action, context) return _convert_action( action, records, context=context, config=config) return action else: record, = records values = record._on_change_args(change) values['id'] = record.id changes = getattr(proxy, button)(values, context) record._set_on_change(changes) def _get_values(self, fields=None): 'Return dictionary values' if fields is None: fields = self._values.keys() values = {} for name in fields: if name in ['id', '_timestamp']: continue definition = self._fields[name] if definition.get('readonly') and definition['type'] != 'one2many': continue values[name] = getattr(self, '__%s_value' % name) # Sending an empty X2Many fields breaks ModelFieldAccess.check if (definition['type'] in {'one2many', 'many2many'} and not values[name]): del values[name] return values @property def _timestamp(self): 'Get _timestamp' return self._values.get('_timestamp') def _get_timestamp(self): 'Return dictionary with timestamps' result = {'%s,%s' % (self.__class__.__name__, self.id): self._timestamp} for field, definition in self._fields.items(): if field not in self._values: continue if definition['type'] in ('one2many', 'many2many'): for record in getattr(self, field): result.update(record._get_timestamp()) return result def _read(self, name): 'Read field' fields = [name] if name in self._values: return loading = self._fields[name]['loading'] if loading == 'eager': fields = [x for x, y in self._fields.items() if y['loading'] == 'eager'] fields.append('_timestamp') self._set(self._proxy.read([self.id], fields, self._context)[0]) def _default_get(self): 'Set default values' fields = list(self._fields.keys()) self._default_set( self._proxy.default_get(fields, False, self._context)) def _set(self, values, _default=False): fieldnames = [] for field, value in values.items(): if '.' in field: continue if ((definition := self._fields.get(field)) and definition['type'] in {'one2many', 'many2many'}): if value and len(value) and isinstance(value[0], int): self._values[field] = value else: Relation = Model.get(definition['relation'], self._config) self._values[field] = records = ModelList( definition, [], self, field) for vals in (value or []): record = Relation() record._set(vals, _default=_default) records.append(record) else: self._values[field] = value fieldnames.append(field) if _default: self._on_change(sorted(fieldnames)) def _default_set(self, values): return self._set(values, _default=True) def _get_eval(self): values = dict((x, getattr(self, '__%s_eval' % x)) for x in self._fields if x != 'id') values['id'] = self.id return values def _get_on_change_values(self, skip=None, fields=None): values = {'id': self.id} if fields: definitions = ((f, self._fields[f]) for f in fields) else: definitions = self._fields.items() for field, definition in definitions: if field == 'id': continue if not fields: if skip and field in skip: continue if (self.id >= 0 and (field not in self._values or field not in self._changed)): continue if definition['type'] == 'one2many': values[field] = [x._get_on_change_values( skip={definition.get('relation_field', '')}) for x in getattr(self, field)] elif (definition['type'] in ('many2one', 'reference') and self._parent_name == definition['name'] and self._parent): values[field] = self._parent._get_on_change_values( skip={self._parent_field_name}) if definition['type'] == 'reference': values[field] = ( self._parent.__class__.__name__, values[field]) else: values[field] = getattr(self, '__%s_eval' % field) return values def _on_change_args(self, args): # Ensure arguments has been read for arg in args: record = self for i in arg.split('.'): if i in record._fields: getattr(record, i) elif i == '_parent_' + record._parent_name: getattr(record, record._parent_name) record = record._parent res = {} values = _EvalEnvironment(self, 'on_change') for arg in args: scope = values for i in arg.split('.'): if i not in scope: break scope = scope[i] else: res[arg] = scope return res def _on_change_set(self, field, value): if (self._fields[field]['type'] in ('one2many', 'many2many') and not isinstance(value, (list, tuple))): if value and value.get('delete'): for record_id in value['delete']: for record in getattr(self, field): if record.id == record_id: getattr(self, field).remove(record, _changed=False) if value and value.get('remove'): for record_id in value['remove']: for i, record in enumerate(getattr(self, field)): if record.id == record_id: getattr(self, field).pop(i, _changed=False) if value and (value.get('add') or value.get('update')): for index, vals in value.get('add', []): group = getattr(self, field) Relation = Model.get( self._fields[field]['relation'], self._config) config = Relation._config id_ = vals.pop('id', None) with config.reset_context(), \ config.set_context(self._context): record = Relation(id=id_, _group=group, _default=False) try: idx = group.index(record) except ValueError: # append without signal if index == -1: list.append(group, record) else: list.insert(group, index, record) else: record = group[idx] group.record_removed.discard(record) group.record_deleted.discard(record) record._set_on_change(vals) for vals in value.get('update', []): if 'id' not in vals: continue for record in getattr(self, field): if record.id == vals['id']: record._set_on_change(vals) elif (self._fields[field]['type'] in ('one2many', 'many2many') and len(value) and not isinstance(value[0], int)): self._values[field] = [] for vals in value: Relation = Model.get( self._fields[field]['relation'], self._config) config = Relation._config records = getattr(self, field) with config.reset_context(), \ config.set_context(records._get_context()): record = Relation(_default=False, **vals) records.append(record) else: self._values[field] = value self._changed.add(field) def _set_on_change(self, values): later = {} for field, value in values.items(): if field not in self._fields: continue if self._fields[field]['type'] in ('one2many', 'many2many'): later[field] = value continue self._on_change_set(field, value) for field, value in later.items(): self._on_change_set(field, value) def _on_change(self, names): 'Call on_change for field' # Import locally to not break installation from proteus.pyson import PYSONDecoder values = {} for name in names: definition = self._fields[name] on_change = definition.get('on_change') if not on_change: continue if isinstance(on_change, str): definition['on_change'] = on_change = PYSONDecoder().decode( on_change) values.update(self._on_change_args(on_change)) if values: values['id'] = self.id context = self._context change = getattr(self._proxy, 'on_change')(values, names, context) self._set_on_change(change) values = {} fieldnames = set(names) to_change = set() later = set() for field, definition in self._fields.items(): on_change_with = definition.get('on_change_with') if not on_change_with: continue if not fieldnames & set(on_change_with): continue if to_change & set(on_change_with): later.add(field) continue to_change.add(field) values.update(self._on_change_args(on_change_with + [field])) if to_change: values['id'] = self.id context = self._context changes = getattr(self._proxy, 'on_change_with')(values, list(to_change), context) self._set_on_change(changes) if later: values = {} for field in later: on_change_with = self._fields[field]['on_change_with'] values.update(self._on_change_args(on_change_with + [field])) values['id'] = self.id context = self._context changes = getattr(self._proxy, 'on_change_with')( values, list(later), context) self._set_on_change(changes) if self._parent: self._parent._changed.add(self._parent_field_name) self._parent._on_change([self._parent_field_name]) def notifications(self): values = self._get_on_change_values() return getattr(self._proxy, 'on_change_notify')(values, self._context) class Wizard(object): 'Wizard class for Tryton wizards' __slots__ = ('name', 'form', 'form_state', 'actions', '_config', '_context', '_proxy', 'session_id', 'start_state', 'end_state', 'states', 'state', 'models', 'action') def __init__(self, name, models=None, action=None, config=None, context=None): if models: assert len(set(type(x) for x in models)) == 1 super().__init__() self.name = name self.form = None self.form_state = None self.actions = [] self._config = config or proteus.config.get_config() self._context = self._config.context if context: self._context.update(context) self._proxy = self._config.get_proxy(name, type='wizard') result = self._proxy.create(self._context) self.session_id, self.start_state, self.end_state = result self.states = [self.start_state] self.models = models self.action = action self.execute(self.start_state) def execute(self, state): assert state in self.states self.state = state while self.state != self.end_state: ctx = self._context.copy() if self.models: ctx['active_id'] = self.models[0].id ctx['active_ids'] = [model.id for model in self.models] ctx['active_model'] = self.models[0].__class__.__name__ elif isinstance(self.models, ModelList): ctx['active_id'] = None ctx['active_ids'] = None ctx['active_model'] = self.models.model_name else: ctx['active_id'] = None ctx['active_ids'] = None ctx['active_model'] = None if self.action: ctx['action_id'] = self.action['id'] else: ctx['action_id'] = None if self.form: data = {self.form_state: self.form._get_on_change_values()} else: data = {} result = self._proxy.execute(self.session_id, data, self.state, ctx) if 'view' in result: view = result['view'] self.form = Model.get( view['fields_view']['model'], self._config)(_default=False) if 'defaults' in view: self.form._default_set(view['defaults']) if 'values' in view: self.form._set(view['values']) self.states = [b['state'] for b in view['buttons']] self.form_state = view['state'] else: self.state = self.end_state self.actions = [] for action in result.get('actions', []): proteus_action = _convert_action( *action, context=self._context, config=self._config) if proteus_action is not None: self.actions.append(proteus_action) if 'view' in result: return if self.state == self.end_state: self._proxy.delete(self.session_id, self._context) if self.models: for record in self.models: record.reload() class Report(object): 'Report class for Tryton reports' __slots__ = ('name', '_config', '_context', '_proxy') def __init__(self, name, config=None, context=None): super().__init__() self.name = name self._config = config or proteus.config.get_config() self._context = self._config.context if context: self._context.update(context) self._proxy = self._config.get_proxy(name, type='report') def execute(self, models=None, data=None): if models: ids = [m.id for m in models] elif data: ids = data.get('ids', []) else: ids = [] if data is None: data = { 'id': ids[0] if ids else None, 'ids': ids, } if models: data['model'] = models[0].__class__.__name__ return self._proxy.execute(ids, data, self._context) def launch_action(xml_id, records, context=None, config=None): if records: assert len({type(x) for x in records}) == 1 if context is None: context = {} if isinstance(xml_id, str): ModelData = Model.get('ir.model.data') if not config: config = proteus.config.get_config() context = config.context action_id = ModelData.get_id(*xml_id.split('.'), context) elif isinstance(xml_id, int): action_id = xml_id Action = Model.get('ir.action') action = Action.get_action_value(action_id, context) return _convert_action(action, records, context=context, config=config) def _convert_action(action, data=None, *, context=None, config=None): if config is None: config = proteus.config.get_config() records = None if data is None: data = {} elif isinstance(data, (list, tuple)): records = data data = { 'model': records[0].__class__.__name__, 'id': records[0].id, 'ids': [r.id for r in records], } else: data = data.copy() if 'type' not in (action or {}): return None data['action_id'] = action['id'] if action['type'] == 'ir.action.act_window': from .pyson import PYSONDecoder action.setdefault('pyson_domain', '[]') ctx = { 'active_model': data.get('model'), 'active_id': data.get('id'), 'active_ids': data.get('ids', []), } ctx.update(config.context) ctx['_user'] = config.user decoder = PYSONDecoder(ctx) action_ctx = decoder.decode(action.get('pyson_context') or '{}') ctx.update(action_ctx) ctx.update(context) action_ctx.update(context) if 'date_format' not in action_ctx: action_ctx['date_format'] = config.context.get( 'locale', {}).get('date', '%x') ctx['context'] = ctx decoder = PYSONDecoder(ctx) domain = decoder.decode(action['pyson_domain']) res_model = action.get('res_model', data.get('res_model')) res_id = action.get('res_id', data.get('res_id')) Model_ = Model.get(res_model, config) config = Model_._config with config.reset_context(), config.set_context(action_ctx): if res_id is None: return Model_.find(domain) elif isinstance(res_id, int): return [Model_(res_id)] else: return [Model_(id_) for id_ in res_id] elif action['type'] == 'ir.action.wizard': kwargs = { 'action': action, 'config': config, 'context': context, } if records is not None: kwargs['models'] = records elif 'model' in data: Model_ = Model.get(data['model'], config) config = Model_._config with config.reset_context(), config.set_context(context): kwargs['models'] = [Model_(id_) for id_ in data.get('ids', [])] return Wizard(action['wiz_name'], **kwargs) elif action['type'] == 'ir.action.report': ActionReport = Report(action['report_name'], context=context) return ActionReport.execute(data=data) elif action['type'] == 'ir.action.url': return action.get('url')