""" Check if the given name is a valid model name.

The _name attribute in osv and osv_memory object is subject to

some restrictions. This function returns True or False whether

the given name is allowed or not.

TODO: this is an approximation. The goal in this approximation

is to disallow uppercase characters (in some places, we quote

table/column names and in other not, which leads to this kind

of errors:

psycopg2.ProgrammingError: relation "xxx" does not exist).

The same restriction should apply to both osv and osv_memory

objects for consistency.

"""

if regex_object_name.match(name) is None:

return False

""" Check whether the given name is a valid PostgreSQL identifier name. """

if not regex_pg_name.match(name):

raise ValidationError("Invalid characters in table name %r" % name)

if len(name) > 63:

"""

Fixes the id fields in import and exports, and splits field paths

on '/'.

:param str fieldname: name of the field to import/export

:return: split field name

:rtype: list of str

"""

fixed_db_id = re.sub(r'([^/])\.id', r'\1/.id', fieldname)

fixed_external_id = re.sub(r'([^/]):id', r'\1/id', fixed_db_id)

""" The metaclass of all model classes.

Its main purpose is to register the models per module.

"""

module_to_models = defaultdict(list)

def __new__(meta, name, bases, attrs):

# this prevents assignment of non-fields on recordsets

attrs.setdefault('__slots__', ())

# this collects the fields defined on the class (via Field.__set_name__())

attrs.setdefault('_field_definitions', [])

if attrs.get('_register', True):

# determine '_module'

if '_module' not in attrs:

module = attrs['__module__']

assert module.startswith('odoo.addons.'), \

f"Invalid import of {module}.{name}, it should start with 'odoo.addons'."

attrs['_module'] = module.split('.')[2]

# determine model '_name' and normalize '_inherits'

inherit = attrs.get('_inherit', ())

if isinstance(inherit, str):

inherit = attrs['_inherit'] = [inherit]

if '_name' not in attrs:

attrs['_name'] = inherit[0] if len(inherit) == 1 else name

return super().__new__(meta, name, bases, attrs)

def __init__(self, name, bases, attrs):

super().__init__(name, bases, attrs)

if '__init__' in attrs and len(inspect.signature(attrs['__init__']).parameters) != 4:

_logger.warning("The method %s.__init__ doesn't match the new signature in module %s", name, attrs.get('__module__'))

if not attrs.get('_register', True):

return

# Remember which models to instantiate for this module.

if self._module:

self.module_to_models[self._module].append(self)

if not self._abstract and self._name not in self._inherit:

# this class defines a model: add magic fields

def add(name, field):

setattr(self, name, field)

field.__set_name__(self, name)

def add_default(name, field):

if name not in attrs:

setattr(self, name, field)

field.__set_name__(self, name)

add('id', fields.Id(automatic=True))

add(self.CONCURRENCY_CHECK_FIELD, fields.Datetime(

string='Last Modified on', automatic=True,

compute='_compute_concurrency_field', compute_sudo=False))

add_default('display_name', fields.Char(

string='Display Name', automatic=True, compute='_compute_display_name'))

if attrs.get('_log_access', self._auto):

add_default('create_uid', fields.Many2one(

'res.users', string='Created by', automatic=True, readonly=True))

add_default('create_date', fields.Datetime(

string='Created on', automatic=True, readonly=True))

add_default('write_uid', fields.Many2one(

'res.users', string='Last Updated by', automatic=True, readonly=True))

add_default('write_date', fields.Datetime(

""" Pseudo-ids for new records, encapsulating an optional origin id (actual

record id) and an optional reference (any value).

"""

__slots__ = ['origin', 'ref']

def __init__(self, origin=None, ref=None):

self.origin = origin

self.ref = ref

def __bool__(self):

return False

def __eq__(self, other):

return isinstance(other, NewId) and (

(self.origin and other.origin and self.origin == other.origin)

or (self.ref and other.ref and self.ref == other.ref)

)

def __hash__(self):

return hash(self.origin or self.ref or id(self))

def __repr__(self):

return (

"<NewId origin=%r>" % self.origin if self.origin else

"<NewId ref=%r>" % self.ref if self.ref else

"<NewId 0x%x>" % id(self)

)

def __str__(self):

if self.origin or self.ref:

id_part = repr(self.origin or self.ref)

else:

id_part = hex(id(self))

""" Return an iterator over the origin ids corresponding to ``ids``.

Actual ids are returned as is, and ids without origin are not returned.

"""

""" A reversible iterable returning the origin ids of a collection of ``ids``. """

__slots__ = ['ids']

def __init__(self, ids):

self.ids = ids

def __iter__(self):

return origin_ids(self.ids)

def __reversed__(self):

""" Return an iterator of unique ids from the concatenation of ``[id0]`` and

``ids``, and of the same kind (all real or all new).

"""

yield id0

seen = {id0}

kind = bool(id0)

for id_ in ids:

if id_ not in seen and bool(id_) == kind:

yield id_

"""Base class for Odoo models.

Odoo models are created by inheriting one of the following:

* :class:`Model` for regular database-persisted models

* :class:`TransientModel` for temporary data, stored in the database but

automatically vacuumed every so often

* :class:`AbstractModel` for abstract super classes meant to be shared by

multiple inheriting models

The system automatically instantiates every model once per database. Those

instances represent the available models on each database, and depend on

which modules are installed on that database. The actual class of each

instance is built from the Python classes that create and inherit from the

corresponding model.

Every model instance is a "recordset", i.e., an ordered collection of

records of the model. Recordsets are returned by methods like

:meth:`~.browse`, :meth:`~.search`, or field accesses. Records have no

explicit representation: a record is represented as a recordset of one

record.

To create a class that should not be instantiated,

the :attr:`~odoo.models.BaseModel._register` attribute may be set to False.

"""

__slots__ = ['env', '_ids', '_prefetch_ids']

_auto = False

"""Whether a database table should be created.

If set to ``False``, override :meth:`~odoo.models.BaseModel.init`

to create the database table.

Automatically defaults to `True` for :class:`Model` and

:class:`TransientModel`, `False` for :class:`AbstractModel`.

.. tip:: To create a model without any table, inherit

from :class:`~odoo.models.AbstractModel`.

"""

_register = False #: registry visibility

_abstract = True

""" Whether the model is *abstract*.

.. seealso:: :class:`AbstractModel`

"""

_transient = False

""" Whether the model is *transient*.

.. seealso:: :class:`TransientModel`

"""

_name = None #: the model name (in dot-notation, module namespace)

_description = None #: the model's informal name

_module = None #: the model's module (in the Odoo sense)

_custom = False #: should be True for custom models only

_inherit = ()

"""Python-inherited models:

:type: str or list(str)

.. note::

* If :attr:`._name` is set, name(s) of parent models to inherit from

* If :attr:`._name` is unset, name of a single model to extend in-place

"""

_inherits = frozendict()

"""dictionary {'parent_model': 'm2o_field'} mapping the _name of the parent business

objects to the names of the corresponding foreign key fields to use::

_inherits = {

'a.model': 'a_field_id',

'b.model': 'b_field_id'

}

implements composition-based inheritance: the new model exposes all

the fields of the inherited models but stores none of them:

the values themselves remain stored on the linked record.

.. warning::

if multiple fields with the same name are defined in the

:attr:`~odoo.models.Model._inherits`-ed models, the inherited field will

correspond to the last one (in the inherits list order).

"""

_table = None #: SQL table name used by model if :attr:`_auto`

_table_query = None #: SQL expression of the table's content (optional)

_sql_constraints = [] #: SQL constraints [(name, sql_def, message)]

_rec_name = None #: field to use for labeling records, default: ``name``

_rec_names_search = None #: fields to consider in ``name_search``

_order = 'id' #: default order field for searching results

_parent_name = 'parent_id' #: the many2one field used as parent field

_parent_store = False

"""set to True to compute parent_path field.

Alongside a :attr:`~.parent_path` field, sets up an indexed storage

of the tree structure of records, to enable faster hierarchical queries

on the records of the current model using the ``child_of`` and

``parent_of`` domain operators.

"""

_active_name = None

"""field to use for active records, automatically set to either ``"active"``

or ``"x_active"``.

"""

_fold_name = 'fold' #: field to determine folded groups in kanban views

_translate = True # False disables translations export for this model (Old API)

_check_company_auto = False

"""On write and create, call ``_check_company`` to ensure companies

consistency on the relational fields having ``check_company=True``

as attribute.

"""

_allow_sudo_commands = True

"""Allow One2many and Many2many Commands targeting this model in an environment using `sudo()` or `with_user()`.

By disabling this flag, security-sensitive models protect themselves

against malicious manipulation of One2many or Many2many fields

through an environment using `sudo` or a more priviledged user.

"""

_depends = frozendict()

"""dependencies of models backed up by SQL views

``{model_name: field_names}``, where ``field_names`` is an iterable.

This is only used to determine the changes to flush to database before

executing ``search()`` or ``read_group()``. It won't be used for cache

invalidation or recomputing fields.

"""

# default values for _transient_vacuum()

_transient_max_count = lazy_classproperty(lambda _: config.get('osv_memory_count_limit'))

"maximum number of transient records, unlimited if ``0``"

_transient_max_hours = lazy_classproperty(lambda _: config.get('transient_age_limit'))

"maximum idle lifetime (in hours), unlimited if ``0``"

CONCURRENCY_CHECK_FIELD = '__last_update'

def _valid_field_parameter(self, field, name):

""" Return whether the given parameter name is valid for the field. """

return name == 'related_sudo'

@api.model

def _add_field(self, name, field):

""" Add the given ``field`` under the given ``name`` in the class """

cls = self.env.registry[self._name]

# add field as an attribute and in cls._fields (for reflection)

if not isinstance(getattr(cls, name, field), Field):

_logger.warning("In model %r, field %r overriding existing value", cls._name, name)

setattr(cls, name, field)

field._toplevel = True

field.__set_name__(cls, name)

cls._fields[name] = field

@api.model

def _pop_field(self, name):

""" Remove the field with the given ``name`` from the model.

This method should only be used for manual fields.

"""

cls = self.env.registry[self._name]

field = cls._fields.pop(name, None)

discardattr(cls, name)

if cls._rec_name == name:

# fixup _rec_name and display_name's dependencies

cls._rec_name = None

if cls.display_name in cls.pool.field_depends:

cls.pool.field_depends[cls.display_name] = tuple(

dep for dep in cls.pool.field_depends[cls.display_name] if dep != name

)

return field

@api.depends(lambda model: ('create_date', 'write_date') if model._log_access else ())

def _compute_concurrency_field(self):

fname = self.CONCURRENCY_CHECK_FIELD

if self._log_access:

for record in self:

record[fname] = record.write_date or record.create_date or Datetime.now()

else:

self[fname] = odoo.fields.Datetime.now()

#

# Goal: try to apply inheritance at the instantiation level and

# put objects in the pool var

#

@classmethod

def _build_model(cls, pool, cr):

""" Instantiate a given model in the registry.

This method creates or extends a "registry" class for the given model.

This "registry" class carries inferred model metadata, and inherits (in

the Python sense) from all classes that define the model, and possibly

other registry classes.

"""

if getattr(cls, '_constraints', None):

_logger.warning("Model attribute '_constraints' is no longer supported, "

"please use @api.constrains on methods instead.")

# Keep links to non-inherited constraints in cls; this is useful for

# instance when exporting translations

cls._local_sql_constraints = cls.__dict__.get('_sql_constraints', [])

# all models except 'base' implicitly inherit from 'base'

name = cls._name

parents = list(cls._inherit)

if name != 'base':

parents.append('base')

# create or retrieve the model's class

if name in parents:

if name not in pool:

raise TypeError("Model %r does not exist in registry." % name)

ModelClass = pool[name]

ModelClass._build_model_check_base(cls)

check_parent = ModelClass._build_model_check_parent

else:

ModelClass = type(name, (cls,), {

'_name': name,

'_register': False,

'_original_module': cls._module,

'_inherit_module': {}, # map parent to introducing module

'_inherit_children': OrderedSet(), # names of children models

'_inherits_children': set(), # names of children models

'_fields': {}, # populated in _setup_base()

})

check_parent = cls._build_model_check_parent

# determine all the classes the model should inherit from

bases = LastOrderedSet([cls])

for parent in parents:

if parent not in pool:

raise TypeError("Model %r inherits from non-existing model %r." % (name, parent))

parent_class = pool[parent]

if parent == name:

for base in parent_class.__base_classes:

bases.add(base)

else:

check_parent(cls, parent_class)

bases.add(parent_class)

ModelClass._inherit_module[parent] = cls._module

parent_class._inherit_children.add(name)

# ModelClass.__bases__ must be assigned those classes; however, this

# operation is quite slow, so we do it once in method _prepare_setup()

ModelClass.__base_classes = tuple(bases)

# determine the attributes of the model's class

ModelClass._build_model_attributes(pool)

check_pg_name(ModelClass._table)

# Transience

if ModelClass._transient:

assert ModelClass._log_access, \

"TransientModels must have log_access turned on, " \

"in order to implement their vacuum policy"

# link the class to the registry, and update the registry

ModelClass.pool = pool

pool[name] = ModelClass

return ModelClass

@classmethod

def _build_model_check_base(model_class, cls):

""" Check whether ``model_class`` can be extended with ``cls``. """

if model_class._abstract and not cls._abstract:

msg = ("%s transforms the abstract model %r into a non-abstract model. "

"That class should either inherit from AbstractModel, or set a different '_name'.")

raise TypeError(msg % (cls, model_class._name))

if model_class._transient != cls._transient:

if model_class._transient:

msg = ("%s transforms the transient model %r into a non-transient model. "

"That class should either inherit from TransientModel, or set a different '_name'.")

else:

msg = ("%s transforms the model %r into a transient model. "

"That class should either inherit from Model, or set a different '_name'.")

raise TypeError(msg % (cls, model_class._name))

@classmethod

def _build_model_check_parent(model_class, cls, parent_class):

""" Check whether ``model_class`` can inherit from ``parent_class``. """

if model_class._abstract and not parent_class._abstract:

msg = ("In %s, the abstract model %r cannot inherit from the non-abstract model %r.")

raise TypeError(msg % (cls, model_class._name, parent_class._name))

@classmethod

def _build_model_attributes(cls, pool):

""" Initialize base model attributes. """

cls._description = cls._name

cls._table = cls._name.replace('.', '_')

cls._log_access = cls._auto

inherits = {}

depends = {}

cls._sql_constraints = {}

for base in reversed(cls.__base_classes):

if is_definition_class(base):

# the following attributes are not taken from registry classes

if cls._name not in base._inherit and not base._description:

_logger.warning("The model %s has no _description", cls._name)

cls._description = base._description or cls._description

cls._table = base._table or cls._table

cls._log_access = getattr(base, '_log_access', cls._log_access)

inherits.update(base._inherits)

for mname, fnames in base._depends.items():

depends.setdefault(mname, []).extend(fnames)

for cons in base._sql_constraints:

cls._sql_constraints[cons[0]] = cons

cls._sql_constraints = list(cls._sql_constraints.values())

# avoid assigning an empty dict to save memory

if inherits:

cls._inherits = inherits

if depends:

cls._depends = depends

# update _inherits_children of parent models

for parent_name in cls._inherits:

pool[parent_name]._inherits_children.add(cls._name)

# recompute attributes of _inherit_children models

for child_name in cls._inherit_children:

child_class = pool[child_name]

child_class._build_model_attributes(pool)

@classmethod

def _init_constraints_onchanges(cls):

# store list of sql constraint qualified names

for (key, _, _) in cls._sql_constraints:

cls.pool._sql_constraints.add(cls._table + '_' + key)

# reset properties memoized on cls

cls._constraint_methods = BaseModel._constraint_methods

cls._ondelete_methods = BaseModel._ondelete_methods

cls._onchange_methods = BaseModel._onchange_methods

@property

def _constraint_methods(self):

""" Return a list of methods implementing Python constraints. """

def is_constraint(func):

return callable(func) and hasattr(func, '_constrains')

def wrap(func, names):

# wrap func into a proxy function with explicit '_constrains'

@api.constrains(*names)

def wrapper(self):

return func(self)

return wrapper

cls = self.env.registry[self._name]

methods = []

for attr, func in getmembers(cls, is_constraint):

if callable(func._constrains):

func = wrap(func, func._constrains(self))

for name in func._constrains:

field = cls._fields.get(name)

if not field:

_logger.warning("method %s.%s: @constrains parameter %r is not a field name", cls._name, attr, name)

elif not (field.store or field.inverse or field.inherited):

_logger.warning("method %s.%s: @constrains parameter %r is not writeable", cls._name, attr, name)

methods.append(func)

# optimization: memoize result on cls, it will not be recomputed

cls._constraint_methods = methods

return methods

@property

def _ondelete_methods(self):

""" Return a list of methods implementing checks before unlinking. """

def is_ondelete(func):

return callable(func) and hasattr(func, '_ondelete')

cls = self.env.registry[self._name]

methods = [func for _, func in getmembers(cls, is_ondelete)]

# optimization: memoize results on cls, it will not be recomputed

cls._ondelete_methods = methods

return methods

@property

def _onchange_methods(self):

""" Return a dictionary mapping field names to onchange methods. """

def is_onchange(func):

return callable(func) and hasattr(func, '_onchange')

# collect onchange methods on the model's class

cls = self.env.registry[self._name]

methods = defaultdict(list)

for attr, func in getmembers(cls, is_onchange):

missing = []

for name in func._onchange:

if name not in cls._fields:

missing.append(name)

methods[name].append(func)

if missing:

_logger.warning(

"@api.onchange%r parameters must be field names -> not valid: %s",

func._onchange, missing

)

# add onchange methods to implement "change_default" on fields

def onchange_default(field, self):

value = field.convert_to_write(self[field.name], self)

condition = "%s=%s" % (field.name, value)

defaults = self.env['ir.default'].get_model_defaults(self._name, condition)

self.update(defaults)

for name, field in cls._fields.items():

if field.change_default:

methods[name].append(functools.partial(onchange_default, field))

# optimization: memoize result on cls, it will not be recomputed

cls._onchange_methods = methods

return methods

def _is_an_ordinary_table(self):

return self.pool.is_an_ordinary_table(self)

def __ensure_xml_id(self, skip=False):

""" Create missing external ids for records in ``self``, and return an

iterator of pairs ``(record, xmlid)`` for the records in ``self``.

:rtype: Iterable[Model, str | None]

"""

if skip:

return ((record, None) for record in self)

if not self:

return iter([])

if not self._is_an_ordinary_table():

raise Exception(

"You can not export the column ID of model %s, because the "

"table %s is not an ordinary table."

% (self._name, self._table))

modname = '__export__'

cr = self.env.cr

cr.execute("""

SELECT res_id, module, name

FROM ir_model_data

WHERE model = %s AND res_id in %s

""", (self._name, tuple(self.ids)))

xids = {

res_id: (module, name)

for res_id, module, name in cr.fetchall()

}

def to_xid(record_id):

(module, name) = xids[record_id]

return ('%s.%s' % (module, name)) if module else name

# create missing xml ids

missing = self.filtered(lambda r: r.id not in xids)

if not missing:

return (

(record, to_xid(record.id))

for record in self

)

xids.update(

(r.id, (modname, '%s_%s_%s' % (

r._table,

r.id,

uuid.uuid4().hex[:8],

)))

for r in missing

)

fields = ['module', 'model', 'name', 'res_id']

# disable eventual async callback / support for the extent of

# the COPY FROM, as these are apparently incompatible

callback = psycopg2.extensions.get_wait_callback()

psycopg2.extensions.set_wait_callback(None)

try:

cr.copy_from(io.StringIO(

u'\n'.join(

u"%s\t%s\t%s\t%d" % (

modname,

record._name,

xids[record.id][1],

record.id,

)

for record in missing

)),

table='ir_model_data',

columns=fields,

)

finally:

psycopg2.extensions.set_wait_callback(callback)

self.env['ir.model.data'].invalidate_model(fields)

return (

(record, to_xid(record.id))

for record in self

)

def _export_rows(self, fields, *, _is_toplevel_call=True):

""" Export fields of the records in ``self``.

:param list fields: list of lists of fields to traverse

:param bool _is_toplevel_call:

used when recursing, avoid using when calling from outside

:return: list of lists of corresponding values

"""

import_compatible = self.env.context.get('import_compat', True)

lines = []

def splittor(rs):

""" Splits the self recordset in batches of 1000 (to avoid

entire-recordset-prefetch-effects) & removes the previous batch

from the cache after it's been iterated in full

"""

for idx in range(0, len(rs), 1000):

sub = rs[idx:idx+1000]

for rec in sub:

yield rec

sub.invalidate_recordset()

if not _is_toplevel_call:

splittor = lambda rs: rs

# memory stable but ends up prefetching 275 fields (???)

for record in splittor(self):

# main line of record, initially empty

current = [''] * len(fields)

lines.append(current)

# list of primary fields followed by secondary field(s)

primary_done = []

# process column by column

for i, path in enumerate(fields):

if not path:

continue

name = path[0]

if name in primary_done:

continue

if name == '.id':

current[i] = str(record.id)

elif name == 'id':

current[i] = (record._name, record.id)

else:

field = record._fields[name]

value = record[name]

# this part could be simpler, but it has to be done this way

# in order to reproduce the former behavior

if not isinstance(value, BaseModel):

current[i] = field.convert_to_export(value, record)

else:

primary_done.append(name)

# recursively export the fields that follow name; use

# 'display_name' where no subfield is exported

fields2 = [(p[1:] or ['display_name'] if p and p[0] == name else [])

for p in fields]

# in import_compat mode, m2m should always be exported as

# a comma-separated list of xids or names in a single cell

if import_compatible and field.type == 'many2many':

index = None

# find out which subfield the user wants & its

# location as we might not get it as the first

# column we encounter

for name in ['id', 'name', 'display_name']:

with contextlib.suppress(ValueError):

index = fields2.index([name])

break

if index is None:

# not found anything, assume we just want the

# name_get in the first column

name = None

index = i

if name == 'id':

xml_ids = [xid for _, xid in value.__ensure_xml_id()]

current[index] = ','.join(xml_ids)

else:

current[index] = field.convert_to_export(value, record)

continue

lines2 = value._export_rows(fields2, _is_toplevel_call=False)

if lines2:

# merge first line with record's main line

for j, val in enumerate(lines2[0]):