fixup! [ADD] util.fields: handle ir.exports model/fields renames/removal

aj-fuentes · aj-fuentes · commit 70cdb944554b · 2024-06-18T10:22:12.000+02:00
diff --git a/src/util/fields.py b/src/util/fields.py
@@ -16,7 +16,7 @@
 
 import psycopg2
 from psycopg2 import sql
-from psycopg2.extras import Json, execute_values
+from psycopg2.extras import Json
 
 try:
     from odoo import release
@@ -40,7 +40,13 @@ def make_index_name(table_name, column_name):
 from .const import ENVIRON
 from .domains import _adapt_one_domain, _replace_path, _valid_path_to, adapt_domains
 from .exceptions import SleepyDeveloperError
-from .helpers import _dashboard_actions, _validate_model, resolve_model_fields_path, table_of_model
+from .helpers import (
+    _dashboard_actions,
+    _remove_export_lines,
+    _validate_model,
+    resolve_model_fields_path,
+    table_of_model,
+)
 from .inherit import for_each_inherit
 from .misc import SelfPrintEvalContext, log_progress, version_gte
 from .orm import env, invalidate
@@ -79,126 +85,6 @@ def make_index_name(table_name, column_name):
 )
 
 
-def _get_resolved_ir_exports(cr, models=None, fields=None):
-    """
-    Return a list of ir.exports.line records which models or fields match the given arguments.
-
-    Export lines can reference nested models through relationship field "paths"
-    (e.g. "partner_id/country_id/name"), therefore these needs to be resolved properly.
-
-    Only one of ``models`` or ``fields`` arguments should be provided.
-
-    :param list[str] models: a list of model names to match in exports
-    :param list[(str, str)] fields: a list of (model, field) tuples to match in exports
-    :return: the resolved field paths parts for each matched export line id
-    :rtype: dict[int, list[FieldsPathPart]]
-
-    :meta private: exclude from online docs
-    """
-    assert bool(models) ^ bool(fields), "One of models or fields must be given, and not both."
-
-    # Get the model fields paths for exports.
-    # When matching fields we can already broadly filter on field names (will be double-checked later).
-    # When matching models we can't exclude anything because we don't know intermediate models.
-    where = ""
-    params = {}
-    if fields:
-        fields = {(model, fields) for model, fields in fields}  # noqa: C416  # make sure set[tuple]
-        where = "WHERE el.name ~ ANY(%(field_names)s)"
-        params["field_names"] = [f[1] for f in fields]
-    cr.execute(
-        """
-        SELECT el.id, e.resource AS model, string_to_array(el.name, '/') AS path
-          FROM ir_exports e
-          JOIN ir_exports_line el ON e.id = el.export_id
-        {where}
-        """.format(where=where),
-        params,
-    )
-    paths_to_line_ids = {}
-    for line_id, model, path in cr.fetchall():
-        paths_to_line_ids.setdefault((model, tuple(path)), set()).add(line_id)
-
-    # Resolve intermediate models for all model fields paths, filter only matching paths parts
-    matching_paths_parts = {}
-    for model, path in paths_to_line_ids:
-        resolved_paths = resolve_model_fields_path(cr, model, path)
-        if fields:
-            matching_parts = [p for p in resolved_paths if (p.field_model, p.field_name) in fields]
-        else:
-            matching_parts = [p for p in resolved_paths if p.field_model in models]
-        if not matching_parts:
-            continue
-        matching_paths_parts[(model, path)] = matching_parts
-
-    # Return the matched parts for each export line id
-    result = {}
-    for (model, path), matching_parts in matching_paths_parts.items():
-        line_ids = paths_to_line_ids.get((model, path))
-        if not line_ids:
-            continue  # wut?
-        for line_id in line_ids:
-            result.setdefault(line_id, []).extend(matching_parts)
-    return result
-
-
-def rename_ir_exports_fields(cr, models_fields_map):
-    """
-    Rename fields references in ir.exports.line records.
-
-    :param dict[str, dict[str, str]] models_fields_map: a dict of models to the fields rename dict,
-        like: `{"model.name": {"old_field": "new_field", ...}, ...}`
-
-    :meta private: exclude from online docs
-    """
-    matching_exports = _get_resolved_ir_exports(
-        cr,
-        fields=[(model, field) for model, fields_map in models_fields_map.items() for field in fields_map],
-    )
-    if not matching_exports:
-        return
-    _logger.debug("Renaming %d export template lines with renamed fields", len(matching_exports))
-    fixed_lines_paths = {}
-    for line_id, resolved_paths in matching_exports.items():
-        for path_part in resolved_paths:
-            assert path_part.field_model in models_fields_map
-            fields_map = models_fields_map[path_part.field_model]
-            assert path_part.field_name in fields_map
-            assert path_part.path[path_part.part_index - 1] == path_part.field_name
-            new_field_name = fields_map[path_part.field_name]
-            fixed_path = fixed_lines_paths.get(line_id, list(path_part.path))
-            fixed_path[path_part.part_index - 1] = new_field_name
-            fixed_lines_paths[line_id] = fixed_path
-    execute_values(
-        cr,
-        """
-        UPDATE ir_exports_line el
-           SET name = v.name
-          FROM (VALUES %s) AS v(id, name)
-         WHERE el.id = v.id
-        """,
-        [(k, "/".join(v)) for k, v in fixed_lines_paths.items()],
-    )
-
-
-def remove_ir_exports_lines(cr, models=None, fields=None):
-    """
-    Delete ir.exports.line records that reference models or fields that are/will be removed.
-
-    Only one of ``models`` or ``fields`` arguments should be provided.
-
-    :param list[str] models: a list of model names to match in exports
-    :param list[(str, str)] fields: a list of (model, field) tuples to match in exports
-
-    :meta private: exclude from online docs
-    """
-    matching_exports = _get_resolved_ir_exports(cr, models=models, fields=fields)
-    if not matching_exports:
-        return
-    _logger.debug("Deleting %d export template lines with removed models/fields", len(matching_exports))
-    cr.execute("DELETE FROM ir_exports_line WHERE id IN %s", [tuple(matching_exports.keys())])
-
-
 def ensure_m2o_func_field_data(cr, src_table, column, dst_table):
     """
     Fix broken m2o relations.
@@ -323,7 +209,7 @@ def clean_context(context):
         )
 
     # ir.exports.line
-    remove_ir_exports_lines(cr, fields=[(model, fieldname)])
+    _remove_export_lines(cr, model, fieldname)
 
     def adapter(leaf, is_or, negated):
         # replace by TRUE_LEAF, unless negated or in a OR operation but not negated
@@ -1195,7 +1081,31 @@ def _update_field_usage_multi(cr, models, old, new, domain_adapter=None, skip_in
 
         # ir.exports.line
         if only_models:
-            rename_ir_exports_fields(cr, {model: {old: new} for model in only_models})
+            cr.execute(
+                """
+                SELECT el.id,
+                       e.resource,
+                       STRING_TO_ARRAY(el.name, '/')
+                  FROM ir_exports_line el
+                  JOIN ir_exports e
+                    ON el.export_id = e.id
+                 WHERE el.name ~ %s
+                """,
+                [r"\y{}\y".format(old)],
+            )
+            fixed_lines_paths = {}
+            for line_id, line_model, line_path in cr.fetchall():
+                new_path = [
+                    new if x.field_name == old and x.field_model in only_models else x.field_name
+                    for x in resolve_model_fields_path(cr, line_model, line_path)
+                ]
+                if len(new_path) == len(line_path) and new_path != line_path:
+                    fixed_lines_paths[line_id] = "/".join(new_path)
+            if fixed_lines_paths:
+                cr.execute(
+                    "UPDATE ir_exports_line SET name = (%s::jsonb)->>(id::text) WHERE id IN %s",
+                    [Json(fixed_lines_paths), tuple(fixed_lines_paths)],
+                )
 
         # mail.alias
         if column_exists(cr, "mail_alias", "alias_defaults"):
diff --git a/src/util/helpers.py b/src/util/helpers.py
@@ -292,3 +292,27 @@ def resolve_model_fields_path(cr, model, path):
         {"model": model, "path": list(path)},
     )
     return [FieldsPathPart(**row) for row in cr.dictfetchall()]
+
+
+def _remove_export_lines(cr, model, field=None):
+    q = """
+    SELECT el.id,
+           e.resource,
+           STRING_TO_ARRAY(el.name, '/')
+      FROM ir_exports_line el
+      JOIN ir_exports e
+        ON el.export_id = e.id
+    """
+    if field:
+        q = cr.mogrify(q + " WHERE el.name ~ %s ", [r"\y{}\y".format(field)]).decode()
+    cr.execute(q)
+    to_rem = [
+        line_id
+        for line_id, line_model, line_path in cr.fetchall()
+        if any(
+            x.field_model == model and (field is None or x.field_name == field)
+            for x in resolve_model_fields_path(cr, line_model, line_path)
+        )
+    ]
+    if to_rem:
+        cr.execute("DELETE FROM ir_exports_line WHERE id IN %s", [tuple(to_rem)])
diff --git a/src/util/models.py b/src/util/models.py
@@ -10,8 +10,8 @@
 import re
 
 from .const import ENVIRON
-from .fields import IMD_FIELD_PATTERN, remove_field, remove_ir_exports_lines
-from .helpers import _ir_values_value, _validate_model, model_of_table, table_of_model
+from .fields import IMD_FIELD_PATTERN, remove_field
+from .helpers import _ir_values_value, _remove_export_lines, _validate_model, model_of_table, table_of_model
 from .indirect_references import indirect_references
 from .inherit import for_each_inherit, inherit_parents
 from .misc import _cached, chunks, log_progress
@@ -130,7 +130,7 @@ def remove_model(cr, model, drop_table=True, ignore_m2m=()):
                 notify = notify or bool(cr.rowcount)
 
     # for ir.exports.line we have to take care of "nested" references in fields "paths"
-    remove_ir_exports_lines(cr, models=[model])
+    _remove_export_lines(cr, model)
 
     _rm_refs(cr, model)
 
