Fix category_id import regression by deriving missing relation info\n\n-

bosd · bosd · commit 8316be6fc81f · 2025-09-15T12:04:12.000+02:00
Add warning when relation_table or relation_field are missing from Odoo
metadata\n- Implement automatic derivation of missing relation info
based on Odoo conventions\n- Refactor relational_import.py to reduce
complexity and fix line length issues\n- Fix mypy type errors in test
files\n- All tests now pass and pre-commit hooks are satisfied\n\nThis
resolves the regression where tags were not being set on partners during
import due to missing relation_table and relation_field in strategy
details.
diff --git a/src/odoo_data_flow/lib/preflight.py b/src/odoo_data_flow/lib/preflight.py
@@ -61,9 +61,18 @@ def _handle_m2m_field(
                 "relation": relation,
             }
     else:
+        # Log a warning when relation information is incomplete
+        log.warning(
+            f"Field '{clean_field_name}' is missing relation_table or relation_field "
+            f"in Odoo metadata. This may cause issues with relational import."
+        )
         # Fallback strategy when relation information is incomplete
+        # Include whatever information we have, but don't set strategy to write_tuple
+        # since we don't have the required fields for it
         strategy_details = {
             "strategy": "write_tuple",
+            "relation_table": relation_table,
+            "relation_field": relation_field,
             "relation": relation,
         }
 
diff --git a/src/odoo_data_flow/lib/relational_import.py b/src/odoo_data_flow/lib/relational_import.py
@@ -88,6 +88,29 @@ def _resolve_related_ids(
         return None
 
 
+def _derive_relation_info(
+    model: str, field: str, related_model_fk: str
+) -> tuple[str, str]:
+    """Derive relation table and field names based on Odoo conventions.
+
+    Args:
+        model: The owning model name
+        field: The field name
+        related_model_fk: The related model name
+
+    Returns:
+        A tuple of (relation_table, relation_field)
+    """
+    # Derive relation table name (typically follows pattern: model1_model2_rel)
+    models = sorted([model, related_model_fk])
+    derived_table = f"{models[0].replace('.', '_')}_{models[1].replace('.', '_')}_rel"
+
+    # Derive the owning model field name (typically model_name_id)
+    derived_field = f"{model.replace('.', '_')}_id"
+
+    return derived_table, derived_field
+
+
 def run_direct_relational_import(
     config: Union[str, dict[str, Any]],
     model: str,
@@ -113,6 +136,21 @@ def run_direct_relational_import(
     owning_model_fk = strategy_details.get("relation_field")
     related_model_fk = strategy_details.get("relation")
 
+    # Try to derive missing information if possible
+    if (not relational_table or not owning_model_fk) and related_model_fk:
+        # Try to derive the relation table and field names
+        derived_table, derived_field = _derive_relation_info(
+            model, field, related_model_fk
+        )
+
+        # Only use derived values if we were missing them
+        if not relational_table:
+            log.info(f"Deriving relation_table for field '{field}': {derived_table}")
+            relational_table = derived_table
+        if not owning_model_fk:
+            log.info(f"Deriving relation_field for field '{field}': {derived_field}")
+            owning_model_fk = derived_field
+
     # If we don't have the required information, we can't proceed with this strategy
     if not relational_table or not owning_model_fk:
         log.error(
@@ -167,6 +205,44 @@ def run_direct_relational_import(
     }
 
 
+def _prepare_link_dataframe(
+    source_df: pl.DataFrame,
+    field: str,
+    owning_df: pl.DataFrame,
+    related_model_df: pl.DataFrame,
+    owning_model_fk: str,
+    related_model_fk: str,
+) -> pl.DataFrame:
+    """Prepare the link table DataFrame for relational imports.
+
+    Args:
+        source_df: The source DataFrame
+        field: The field name
+        owning_df: DataFrame with owning model IDs
+        related_model_df: DataFrame with related model IDs
+        owning_model_fk: The owning model foreign key field name
+        related_model_fk: The related model name
+
+    Returns:
+        The prepared link DataFrame
+    """
+    # Create the link table DataFrame
+    link_df = source_df.select(["id", field]).rename({"id": "external_id"})
+    link_df = link_df.with_columns(pl.col(field).str.split(",")).explode(field)
+
+    # Join to get DB IDs for the owning model
+    link_df = link_df.join(owning_df, on="external_id", how="inner").rename(
+        {"db_id": owning_model_fk}
+    )
+
+    # Join to get DB IDs for the related model
+    link_df = link_df.join(
+        related_model_df.rename({"external_id": field}), on=field, how="inner"
+    ).rename({"db_id": f"{related_model_fk}/id"})
+
+    return link_df
+
+
 def run_write_tuple_import(
     config: Union[str, dict[str, Any]],
     model: str,
@@ -192,7 +268,23 @@ def run_write_tuple_import(
     owning_model_fk = strategy_details.get("relation_field")
     related_model_fk = strategy_details.get("relation")
 
-    # If we don't have the required information, we can't proceed with this strategy
+    # Try to derive missing information if possible
+    if (not relational_table or not owning_model_fk) and related_model_fk:
+        # Try to derive the relation table and field names
+        derived_table, derived_field = _derive_relation_info(
+            model, field, related_model_fk
+        )
+
+        # Only use derived values if we were missing them
+        if not relational_table:
+            log.info(f"Deriving relation_table for field '{field}': {derived_table}")
+            relational_table = derived_table
+        if not owning_model_fk:
+            log.info(f"Deriving relation_field for field '{field}': {derived_field}")
+            owning_model_fk = derived_field
+
+    # If we still don't have the required information, we can't proceed
+    # with this strategy
     if not relational_table or not owning_model_fk:
         log.error(
             f"Cannot run write tuple import for field '{field}': "
@@ -219,20 +311,57 @@ def run_write_tuple_import(
         return False
 
     # 3. Create the link table DataFrame
-    link_df = source_df.select(["id", field]).rename({"id": "external_id"})
-    link_df = link_df.with_columns(pl.col(field).str.split(",")).explode(field)
+    link_df = _prepare_link_dataframe(
+        source_df, field, owning_df, related_model_df, owning_model_fk, related_model_fk
+    )
 
-    # Join to get DB IDs for the owning model
-    link_df = link_df.join(owning_df, on="external_id", how="inner").rename(
-        {"db_id": owning_model_fk}
+    # 4. Create records in the relational table
+    return _create_relational_records(
+        config,
+        model,
+        field,
+        relational_table,
+        owning_model_fk,
+        related_model_fk,
+        link_df,
+        owning_df,
+        related_model_df,
+        original_filename,
+        batch_size,
     )
 
-    # Join to get DB IDs for the related model
-    link_df = link_df.join(
-        related_model_df.rename({"external_id": field}), on=field, how="inner"
-    ).rename({"db_id": f"{related_model_fk}/id"})
 
-    # 4. Create records in the relational table
+def _create_relational_records(
+    config: Union[str, dict[str, Any]],
+    model: str,
+    field: str,
+    relational_table: str,
+    owning_model_fk: str,
+    related_model_fk: str,
+    link_df: pl.DataFrame,
+    owning_df: pl.DataFrame,
+    related_model_df: pl.DataFrame,
+    original_filename: str,
+    batch_size: int,
+) -> bool:
+    """Create records in the relational table.
+
+    Args:
+        config: Configuration for the connection
+        model: The model name
+        field: The field name
+        relational_table: The relational table name
+        owning_model_fk: The owning model foreign key field name
+        related_model_fk: The related model name
+        link_df: The link DataFrame
+        owning_df: DataFrame with owning model IDs
+        related_model_df: DataFrame with related model IDs
+        original_filename: The original filename
+        batch_size: The batch size for processing
+
+    Returns:
+        True if successful, False otherwise
+    """
     if isinstance(config, dict):
         connection = conf_lib.get_connection_from_dict(config)
     else:
@@ -241,8 +370,8 @@ def run_write_tuple_import(
 
     # We need to map back to the original external IDs for failure reporting
     # This is a bit heavy, but necessary for accurate error logs.
-    original_links_df = source_df.select(["id", field]).rename(
-        {"id": "parent_external_id"}
+    original_links_df = link_df.select(["external_id", field]).rename(
+        {"external_id": "parent_external_id"}
     )
     original_links_df = original_links_df.with_columns(
         pl.col(field).str.split(",")
diff --git a/tests/test_m2m_missing_relation_info.py b/tests/test_m2m_missing_relation_info.py
@@ -0,0 +1,159 @@
+"""Tests for handling many2many fields with missing relation information."""
+
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import polars as pl
+
+from odoo_data_flow.enums import PreflightMode
+from odoo_data_flow.lib import preflight, relational_import
+
+
+@patch("odoo_data_flow.lib.preflight.pl.read_csv")
+@patch("odoo_data_flow.lib.preflight.conf_lib.get_connection_from_config")
+def test_handle_m2m_field_missing_relation_info(
+    mock_conf_lib: MagicMock,
+    mock_polars_read_csv: MagicMock,
+    tmp_path: Path,
+) -> None:
+    """Verify that _handle_m2m_field works when relation info is missing."""
+    mock_df_header = MagicMock()
+    mock_df_header.columns = ["id", "name", "category_id"]
+
+    # Setup a more robust mock for the chained Polars calls
+    mock_df_data = MagicMock()
+    (
+        mock_df_data.lazy.return_value.select.return_value.select.return_value.sum.return_value.collect.return_value.item.return_value
+    ) = 100
+    mock_polars_read_csv.side_effect = [mock_df_header, mock_df_data]
+
+    mock_model = mock_conf_lib.return_value.get_model.return_value
+    mock_model.fields_get.return_value = {
+        "id": {"type": "integer"},
+        "name": {"type": "char"},
+        "category_id": {
+            "type": "many2many",
+            "relation": "res.partner.category",
+            # Missing relation_table and relation_field
+        },
+    }
+    import_plan: dict[str, Any] = {}
+    result = preflight.deferral_and_strategy_check(
+        preflight_mode=PreflightMode.NORMAL,
+        model="res.partner",
+        filename="file.csv",
+        config="",
+        import_plan=import_plan,
+    )
+    assert result is True
+    assert "category_id" in import_plan["deferred_fields"]
+    assert import_plan["strategies"]["category_id"]["strategy"] == "write_tuple"
+    # Should include relation info even when missing from Odoo metadata
+    assert "relation" in import_plan["strategies"]["category_id"]
+    # Should include None values for missing fields
+    assert import_plan["strategies"]["category_id"]["relation_table"] is None
+    assert import_plan["strategies"]["category_id"]["relation_field"] is None
+
+
+@patch("odoo_data_flow.lib.relational_import.conf_lib.get_connection_from_config")
+@patch("odoo_data_flow.lib.relational_import._resolve_related_ids")
+def test_run_write_tuple_import_derives_missing_info(
+    mock_resolve_ids: MagicMock,
+    mock_get_conn: MagicMock,
+) -> None:
+    """Verify that run_write_tuple_import derives missing relation info."""
+    # Arrange
+    source_df = pl.DataFrame(
+        {
+            "id": ["p1", "p2"],
+            "name": ["Partner 1", "Partner 2"],
+            "category_id": ["cat1,cat2", "cat2,cat3"],
+        }
+    )
+    mock_resolve_ids.return_value = pl.DataFrame(
+        {"external_id": ["cat1", "cat2", "cat3"], "db_id": [11, 12, 13]}
+    )
+    mock_rel_model = MagicMock()
+    mock_get_conn.return_value.get_model.return_value = mock_rel_model
+
+    # Strategy details with missing relation_table and relation_field
+    strategy_details = {
+        "relation_table": None,  # Missing
+        "relation_field": None,  # Missing
+        "relation": "res.partner.category",
+    }
+    id_map = {"p1": 1, "p2": 2}
+
+    # Act
+    result = relational_import.run_write_tuple_import(
+        "dummy.conf",
+        "res.partner",
+        "category_id",
+        strategy_details,
+        source_df,
+        id_map,
+        1,
+        10,
+        MagicMock(),  # progress
+        MagicMock(),  # task_id
+        "source.csv",
+    )
+
+    # Assert
+    # Should succeed because we derive the missing information
+    assert result is True
+    # Should have called create on the derived relation table
+    mock_get_conn.return_value.get_model.assert_called()
+    mock_rel_model.create.assert_called()
+
+
+@patch("odoo_data_flow.lib.relational_import.conf_lib.get_connection_from_config")
+@patch("odoo_data_flow.lib.relational_import._resolve_related_ids")
+def test_run_direct_relational_import_derives_missing_info(
+    mock_resolve_ids: MagicMock,
+    mock_get_conn: MagicMock,
+) -> None:
+    """Verify that run_direct_relational_import derives missing relation info."""
+    # Arrange
+    source_df = pl.DataFrame(
+        {
+            "id": ["p1", "p2"],
+            "name": ["Partner 1", "Partner 2"],
+            "category_id": ["cat1,cat2", "cat2,cat3"],
+        }
+    )
+    mock_resolve_ids.return_value = pl.DataFrame(
+        {"external_id": ["cat1", "cat2", "cat3"], "db_id": [11, 12, 13]}
+    )
+
+    # Strategy details with missing relation_table and relation_field
+    strategy_details = {
+        "relation_table": None,  # Missing
+        "relation_field": None,  # Missing
+        "relation": "res.partner.category",
+    }
+    id_map = {"p1": 1, "p2": 2}
+
+    # Act
+    result = relational_import.run_direct_relational_import(
+        "dummy.conf",
+        "res.partner",
+        "category_id",
+        strategy_details,
+        source_df,
+        id_map,
+        1,
+        10,
+        MagicMock(),  # progress
+        MagicMock(),  # task_id
+        "source.csv",
+    )
+
+    # Assert
+    # Should succeed because we derive the missing information
+    assert isinstance(result, dict)
+    # Should have derived the relation table name
+    assert "res_partner_res_partner_category_rel" in result["model"]
+    # Should have derived the relation field name
+    assert "res_partner_id" in result["unique_id_field"]
