fix: keyerror: relation_table

Author: bosd

src/odoo_data_flow/lib/preflight.py

@@ -319,19 +319,31 @@ def _plan_deferrals_and_strategies(
                     .collect()
                     .item()
                 )
-                if relation_count >= 500:
-                    strategies[clean_field_name] = {
-                        "strategy": "direct_relational_import",
-                        "relation_table": field_info["relation_table"],
-                        "relation_field": field_info["relation_field"],
-                        "relation": field_info["relation"],
-                    }
+                # Check if required keys exist for many2many fields
+                relation_table = field_info.get("relation_table")
+                relation_field = field_info.get("relation_field")
+                relation = field_info.get("relation")
+                
+                if relation_table and relation_field:
+                    if relation_count >= 500:
+                        strategies[clean_field_name] = {
+                            "strategy": "direct_relational_import",
+                            "relation_table": relation_table,
+                            "relation_field": relation_field,
+                            "relation": relation,
+                        }
+                    else:
+                        strategies[clean_field_name] = {
+                            "strategy": "write_tuple",
+                            "relation_table": relation_table,
+                            "relation_field": relation_field,
+                            "relation": relation,
+                        }
                 else:
+                    # Fallback strategy when relation information is incomplete
                     strategies[clean_field_name] = {
                         "strategy": "write_tuple",
-                        "relation_table": field_info["relation_table"],
-                        "relation_field": field_info["relation_field"],
-                        "relation": field_info["relation"],
+                        "relation": relation,
                     }
             elif is_o2m:
                 deferrable_fields.append(clean_field_name)

src/odoo_data_flow/lib/relational_import.py

@@ -107,9 +107,19 @@ def run_direct_relational_import(
         description=f"Pass 2/2: Updating relations for [bold]{field}[/bold]",
     )
     log.info(f"Running 'Direct Relational Import' for field '{field}'...")
-    relational_table = strategy_details["relation_table"]
-    owning_model_fk = strategy_details["relation_field"]
-    related_model_fk = strategy_details["relation"]
+    
+    # Check if required keys exist
+    relational_table = strategy_details.get("relation_table")
+    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
+    if not relational_table or not owning_model_fk:
+        log.error(
+            f"Cannot run direct relational import for field '{field}': "
+            f"Missing relation_table or relation_field in strategy details."
+        )
+        return None
 
     # 1. Prepare the owning model's IDs
     owning_df = pl.DataFrame({"external_id": id_map.keys(), "db_id": id_map.values()})
@@ -170,9 +180,19 @@ def run_write_tuple_import(
         description=f"Pass 2/2: Updating relations for [bold]{field}[/bold]",
     )
     log.info(f"Running 'Write Tuple' for field '{field}'...")
-    relational_table = strategy_details["relation_table"]
-    owning_model_fk = strategy_details["relation_field"]
-    related_model_fk = strategy_details["relation"]
+    
+    # Check if required keys exist
+    relational_table = strategy_details.get("relation_table")
+    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
+    if not relational_table or not owning_model_fk:
+        log.error(
+            f"Cannot run write tuple import for field '{field}': "
+            f"Missing relation_table or relation_field in strategy details."
+        )
+        return False
 
     # 1. Prepare the owning model's IDs
     owning_df = pl.DataFrame({"external_id": id_map.keys(), "db_id": id_map.values()})

tests/test_preflight.py

@@ -398,6 +398,44 @@ def test_direct_relational_import_strategy_for_large_volumes(
             == "direct_relational_import"
         )
 
+    def test_write_tuple_strategy_when_missing_relation_info(
+        self, mock_polars_read_csv: MagicMock, mock_conf_lib: MagicMock
+    ) -> None:
+        """Verify 'write_tuple' is chosen 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 not have relation_table or relation_field in strategy
+        assert "relation" in import_plan["strategies"]["category_id"]
+
     def test_write_tuple_strategy_for_small_volumes(
         self, mock_polars_read_csv: MagicMock, mock_conf_lib: MagicMock
     ) -> None: