[REF] 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 <[email protected]>

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,22 @@ 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", "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 +686,30 @@ 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):
+        from odoo.addons.base.maintenance.migrations.util.helpers import FieldsPathPart, resolve_model_fields_path
+
+        cr = self.env.cr
+
+        # test with provided paths
+        model, path = "res.currency", ["rate_ids", "company_id", "user_ids", "partner_id"]
+        expected_result = [
+            FieldsPathPart(model, path, 1, "res.currency", "rate_ids", "res.currency.rate"),
+            FieldsPathPart(model, path, 2, "res.currency.rate", "company_id", "res.company"),
+            FieldsPathPart(model, path, 3, "res.company", "user_ids", "res.users"),
+            FieldsPathPart(model, path, 4, "res.users", "partner_id", "res.partner"),
+        ]
+        result = resolve_model_fields_path(cr, model, path)
+        self.assertEqual(result, expected_result)
+
+        model, path = "res.users", ("partner_id", "removed_field", "user_id")
+        expected_result = [
+            FieldsPathPart(model, list(path), 1, "res.users", "partner_id", "res.partner"),
+            FieldsPathPart(model, list(path), 2, "res.partner", "removed_field", None),
+        ]
+        result = 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,16 @@ 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 not resolved_parts:
+        return None
+    last_part = resolved_parts[-1]
+    if last_part.part_index != len(last_part.path):
+        return None
+    return last_part.relation_model  # could be 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,82 @@ def _get_theme_models():
         "theme.website.menu": "website.menu",
         "theme.ir.attachment": "ir.attachment",
     }
+
+
+FieldsPathPart = namedtuple(
+    "FieldsPathPart",
+    "model path part_index field_model field_name relation_model",
+)
+FieldsPathPart.__doc__ = """
+Encapsulate information about a field within a fields path.
+
+:param str model: model to resolve the fields ``path`` from
+:param typing.Sequence[str] path: fields path starting from ``model``
+:param int part_index: index of this field in ``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 the field in ``path`` (in the same order).
+
+    .. example::
+
+       To get the information about the fields path ``user_ids.partner_id.title_id``
+       from model `res.users`, we can call this function as
+
+       .. code-block:: python
+
+          resolve_model_fields_path(cr, "res.users", "user_ids.partner_id.title_id".split("."))
+
+    :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 p.model                       AS model,
+                      p.path                        AS path,
+                      1                             AS part_index,
+                      p.model                       AS field_model,
+                      p.path[1]                     AS field_name,
+                      imf.relation                  AS relation_model
+                 FROM (VALUES (%(model)s, %(path)s)) p(model, path)
+            LEFT JOIN ir_model_fields imf
+                   ON imf.model = p.model
+                  AND imf.name = p.path[1]
+
+            UNION ALL
+
+            -- recursive term
+               SELECT rfp.model,
+                      rfp.path,
+                      rfp.part_index + 1            AS part_index,
+                      rfp.relation_model            AS field_model,
+                      rfp.path[rfp.part_index + 1]  AS field_name,
+                      rimf.relation                 AS relation_model
+                 FROM resolved_fields_path rfp
+            LEFT 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
+                  AND rfp.relation_model IS NOT NULL
+        )
+        SELECT * FROM resolved_fields_path
+        ORDER BY model, path, part_index
+        """,
+        {"model": model, "path": list(path)},
+    )
+    return [FieldsPathPart(**row) for row in cr.dictfetchall()]