[IMP] util.helpers: introduce new resolve_model_fields_path helper

It is a common occurrence to have models metadata values that use a
"path of fields" approach (e.g. fields depends, domains, server actions,
import/export templates, etc.) and to effectively resolve those with all
the intermediate model+fields references of the path parts, either a
for loop is used in python, issuing multple queries, or a recursive CTE
query that does that resolution entirely within PostgreSQL.

In this commit a new `resolve_model_fields_path` helper is introduced
using a recursive CTE to replace some older code using the python-loop
approach.
An additional `FieldsPathPart` named tuple type is added to represent
information of the resolved part of a fields path, and the helper will
return a list of these for callers to then act upon.

Co-authored-by: Alvaro Fuentes <afu@odoo.com>

Author: andreabak

src/base/tests/test_util.py

@@ -19,7 +19,7 @@
 
 from odoo.addons.base.maintenance.migrations import util
 from odoo.addons.base.maintenance.migrations.testing import UnitTestCase, parametrize
-from odoo.addons.base.maintenance.migrations.util.domains import _adapt_one_domain
+from odoo.addons.base.maintenance.migrations.util.domains import _adapt_one_domain, _model_of_path
 from odoo.addons.base.maintenance.migrations.util.exceptions import MigrationError
 
 
@@ -42,6 +42,24 @@ def test_adapt_renamed_field(self):
 
         self.assertEqual(match_domain, new_domain)
 
+    @parametrize(
+        [
+            ("res.currency", [], "res.currency"),
+            ("res.currency", ["rate_ids"], "res.currency.rate"),
+            ("res.currency", ("rate_ids", "company_id"), "res.company"),
+            ("res.currency", ["rate_ids", "company_id", "user_ids"], "res.users"),
+            ("res.currency", ("rate_ids", "company_id", "user_ids", "partner_id"), "res.partner"),
+            ("res.users", ["partner_id"], "res.partner"),
+            ("res.users", ["nonexistent_field"], None),
+            ("res.users", ("partner_id", "active"), None),
+            ("res.users", ("partner_id", "active", "name"), None),
+            ("res.users", ("partner_id", "removed_field"), None),
+        ]
+    )
+    def test_model_of_path(self, model, path, expected):
+        cr = self.env.cr
+        self.assertEqual(_model_of_path(cr, model, path), expected)
+
     def test_change_no_leaf(self):
         # testing plan: updata path of a domain where the last element is not changed
 
@@ -670,6 +688,25 @@ def test_model_table_convertion(self):
             self.assertEqual(table, self.env[model]._table)
             self.assertEqual(util.model_of_table(cr, table), model)
 
+    def test_resolve_model_fields_path(self):
+        cr = self.env.cr
+
+        # test with provided paths
+        model, path = "res.currency", ["rate_ids", "company_id", "user_ids", "partner_id"]
+        expected_result = [
+            util.FieldsPathPart("res.currency", "rate_ids", "res.currency.rate"),
+            util.FieldsPathPart("res.currency.rate", "company_id", "res.company"),
+            util.FieldsPathPart("res.company", "user_ids", "res.users"),
+            util.FieldsPathPart("res.users", "partner_id", "res.partner"),
+        ]
+        result = util.resolve_model_fields_path(cr, model, path)
+        self.assertEqual(result, expected_result)
+
+        model, path = "res.users", ("partner_id", "removed_field", "user_id")
+        expected_result = [util.FieldsPathPart("res.users", "partner_id", "res.partner")]
+        result = util.resolve_model_fields_path(cr, model, path)
+        self.assertEqual(result, expected_result)
+
 
 @unittest.skipIf(
     util.version_gte("saas~17.1"),

src/util/domains.py

@@ -33,7 +33,7 @@
     from openerp.tools.safe_eval import safe_eval
 
 from .const import NEARLYWARN
-from .helpers import _dashboard_actions, _validate_model
+from .helpers import _dashboard_actions, _validate_model, resolve_model_fields_path
 from .inherit import for_each_inherit
 from .misc import SelfPrintEvalContext
 from .pg import column_exists, get_value_or_en_translation, table_exists
@@ -160,21 +160,13 @@ def _get_domain_fields(cr):
 
 
 def _model_of_path(cr, model, path):
-    for field in path:
-        cr.execute(
-            """
-            SELECT relation
-              FROM ir_model_fields
-             WHERE model = %s
-               AND name = %s
-        """,
-            [model, field],
-        )
-        if not cr.rowcount:
-            return None
-        [model] = cr.fetchone()
-
-    return model
+    if not path:
+        return model
+    path = tuple(path)
+    resolved_parts = resolve_model_fields_path(cr, model, path)
+    if len(resolved_parts) == len(path):
+        return resolved_parts[-1].relation_model
+    return None
 
 
 def _valid_path_to(cr, path, from_, to):

src/util/helpers.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 import logging
 import os
+from collections import namedtuple
 
 import lxml
 
@@ -214,3 +215,103 @@ def _get_theme_models():
         "theme.website.menu": "website.menu",
         "theme.ir.attachment": "ir.attachment",
     }
+
+
+FieldsPathPart = namedtuple("FieldsPathPart", "field_model field_name relation_model")
+FieldsPathPart.__doc__ = """
+Encapsulate information about a field within a fields path.
+
+:param str field_model: model of the field
+:param str field_name: name of the field
+:param str relation_model: target model of the field, if relational, otherwise ``None``
+"""
+for _f in FieldsPathPart._fields:
+    getattr(FieldsPathPart, _f).__doc__ = None
+
+
+def resolve_model_fields_path(cr, model, path):
+    """
+    Resolve model fields paths.
+
+    This function returns a list of :class:`~odoo.upgrade.util.helpers.FieldsPathPart`
+    where each item describes a field in ``path`` (in the same order). The returned list
+    could be shorter than the original ``path`` due to a missing field or model, or
+    because there is a non-relational field in the path. The only non-relational field
+    allowed in a fields path is the last one, in which case the returned list has the same
+    length as the input ``path``.
+
+    .. example::
+
+       .. code-block:: python
+
+          >>> util.resolve_model_fields_path(cr, "res.partner", "user_ids.partner_id.title".split("."))
+          [FieldsPathPart(field_model='res.partner', field_name='user_ids', relation_model='res.users'),
+          FieldsPathPart(field_model='res.users', field_name='partner_id', relation_model='res.partner'),
+          FieldsPathPart(field_model='res.partner', field_name='title', relation_model='res.partner.title')]
+
+       Last field is not relational:
+
+       .. code-block:: python
+
+          >>> resolve_model_fields_path(cr, "res.partner", "user_ids.active".split("."))
+          [FieldsPathPart(field_model='res.partner', field_name='user_ids', relation_model='res.users'),
+          FieldsPathPart(field_model='res.users', field_name='active', relation_model=None)]
+
+       The path is wrong, it uses a non-relational field:
+
+       .. code-block:: python
+
+          >>> resolve_model_fields_path(cr, "res.partner", "user_ids.active.name".split("."))
+          [FieldsPathPart(field_model='res.partner', field_name='user_ids', relation_model='res.users'),
+          FieldsPathPart(field_model='res.users', field_name='active', relation_model=None)]
+
+       The path is broken, it uses a non-exiting field:
+
+       .. code-block:: python
+
+          >>> resolve_model_fields_path(cr, "res.partner", "user_ids.non_existing_id.active".split("."))
+          [FieldsPathPart(field_model='res.partner', field_name='user_ids', relation_model='res.users')]
+
+    :param str model: model to resolve the fields ``path`` from
+    :param typing.Sequence[str] path: fields path starting from ``model``
+    :return: list of resolved fields path parts
+    :rtype: list(:class:`~odoo.upgrade.util.helpers.FieldsPathPart`)
+    """
+    path = list(path)
+    cr.execute(
+        """
+        WITH RECURSIVE resolved_fields_path AS (
+            -- non-recursive term
+               SELECT imf.model AS field_model,
+                      imf.name AS field_name,
+                      imf.relation AS relation_model,
+                      p.path AS path,
+                      1 AS part_index
+                 FROM (VALUES (%(model)s, %(path)s)) p(model, path)
+                 JOIN ir_model_fields imf
+                   ON imf.model = p.model
+                  AND imf.name = p.path[1]
+
+            UNION ALL
+
+            -- recursive term
+               SELECT rimf.model AS field_model,
+                      rimf.name AS field_name,
+                      rimf.relation AS relation_model,
+                      rfp.path AS path,
+                      rfp.part_index + 1 AS part_index
+                 FROM resolved_fields_path rfp
+                 JOIN ir_model_fields rimf
+                   ON rimf.model = rfp.relation_model
+                  AND rimf.name = rfp.path[rfp.part_index + 1]
+                WHERE cardinality(rfp.path) > rfp.part_index
+        )
+        SELECT field_model,
+               field_name,
+               relation_model
+          FROM resolved_fields_path
+         ORDER BY part_index
+        """,
+        {"model": model, "path": list(path)},
+    )
+    return [FieldsPathPart(**row) for row in cr.dictfetchall()]