[14.0][IMP] product_import: Import products asynchronously

product_import/__manifest__.py

@@ -13,9 +13,12 @@
         "stock",
         # OCA/edi
         "base_business_document_import",
+        # OCA/queue
+        "queue_job",
     ],
     "data": [
         "security/ir.model.access.csv",
         "wizard/product_import_view.xml",
+        "data/job_function.xml",
     ],
 }

product_import/data/job_function.xml

@@ -0,0 +1,14 @@
+<odoo noupdate="1">
+
+    <record id="channel_product_import" model="queue.job.channel">
+        <field name="name">product_import</field>
+        <field name="parent_id" ref="queue_job.channel_root" />
+    </record>
+
+    <record id="job_create_update_products" model="queue.job.function">
+        <field name="model_id" ref="model_product_import" />
+        <field name="method">_create_update_products</field>
+        <field name="channel_id" ref="channel_product_import" />
+    </record>
+
+</odoo>

product_import/tests/common.py

@@ -11,11 +11,12 @@ class TestCommon(SavepointCase):
     def setUpClass(cls):
         super().setUpClass()
         cls.env = cls.env(context=dict(cls.env.context, tracking_disable=True))
-        cls.wiz_model = cls.env["product.import"]
+        # Execute directly, no job
+        cls.wiz_model = cls.env["product.import"].with_context(queue_job__no_delay=True)
         cls.supplier = cls.env["res.partner"].create({"name": "Catalogue Vendor"})
 
-    def _mock(self, method_name):
-        return mock.patch.object(type(self.wiz_model), method_name)
+    def _mock(self, method_name, **kw):
+        return mock.patch.object(type(self.wiz_model), method_name, **kw)
 
     @property
     def wiz_form(self):

product_import/tests/test_product_import.py

@@ -88,9 +88,15 @@ def test_get_company_id(self):
 
     def test_product_import(self):
         # product.product
-        products = self.wiz_model._create_products(
-            self.parsed_catalog, seller=self.supplier
+        product_obj = self.env["product.product"].with_context(active_test=False)
+        existing = product_obj.search([], order="id")
+
+        wiz = self.wiz_model.create(
+            {"product_file": b"", "product_filename": "test_import.xml"}
         )
+        with self._mock("parse_product_catalogue", return_value=self.parsed_catalog):
+            wiz.import_button()
+        products = product_obj.search([], order="id") - existing
         self.assertEqual(len(products), 3)
         for product, parsed in zip(products, PARSED_CATALOG["products"]):
 

product_import/wizard/product_import.py

@@ -4,12 +4,15 @@
 import logging
 from base64 import b64decode, b64encode
 from datetime import date, timedelta
+from itertools import zip_longest
 
 from lxml import etree
 
 from odoo import _, api, fields, models
 from odoo.exceptions import UserError
 
+CHUNK_SIZE = 40
+
 logger = logging.getLogger(__name__)
 
 
@@ -197,14 +200,46 @@
         return product_vals
 
     @api.model
-    def create_product(self, parsed_product, chatter_msg, seller=None):
-        product_vals = self._prepare_product(parsed_product, chatter_msg, seller=seller)
-        if not product_vals:
-            return False
+    def _create_update_products(self, products, seller_id):
+        """Create / Update a product.
+
+        This method is called from a queue job.
+        """
+
+        seller = self.env["res.partner"].browse(seller_id)
+
+        log_msgs = []
+        for parsed_product in products:
+            product_vals = self._prepare_product(
+                parsed_product, log_msgs, seller=seller
+            )
+            if product_vals:
+                msg = self._create_update_product(product_vals)
+                log_msgs.append(msg)
+
+        return "\n".join(log_msgs)
+
+    @api.model
+    def _create_update_product(self, product_vals):
+        """Create / Update a product.
+
+        This method is called from a queue job.
+        """
+        chatter_msg = []
+
         product = product_vals.pop("recordset", None)
         if product:
+            supplierinfo = product_vals.pop("seller_ids", ())
+            supplierinfo_obj = self.env["product.supplierinfo"]
+            for (command, line_id, values) in supplierinfo:
+                if command == 1:
+                    supplierinfo_obj.browse(line_id).write(values)
+                elif command == 0:
+                    supplierinfo_obj.create({**values, "product_id": product.id})
+                else:
+                    raise RuntimeError(f"Command {command} not supported")
             product.write(product_vals)
-            logger.info("Product %d updated", product.id)
+            logger.debug("Product %s updated", product.default_code)
         else:
             product_active = product_vals.pop("active")
             product = self.env["product.product"].create(product_vals)
@@ -213,33 +248,33 @@
                 # all characteristics into product.template
                 product.flush()
                 product.action_archive()
-            logger.info("Product %d created", product.id)
-        return product
+            logger.debug("Product %s created", product.default_code)
 
-    @api.model
-    def _create_products(self, catalogue, seller, filename=None):
-        products = self.env["product.product"].browse()
-        for product in catalogue.get("products"):
-            record = self.create_product(
-                product,
-                catalogue["chatter_msg"],
-                seller=seller,
-            )
-            if record:
-                products |= record
-        self._bdimport.post_create_or_update(catalogue, seller, doc_filename=filename)
-        logger.info("Products updated for vendor %d", seller.id)
-        return products
+        log_msg = f"Product created/updated {product.id}\n" + "\n".join(chatter_msg)
+        return log_msg
 
-    def import_button(self):
+    def import_button(self, chunk_size=CHUNK_SIZE):
         self.ensure_one()
         file_content = b64decode(self.product_file)
         catalogue = self.parse_product_catalogue(file_content, self.product_filename)
         if not catalogue.get("products"):
             raise UserError(_("This catalogue doesn't have any product!"))
         company_id = self._get_company_id(catalogue)
         seller = self._get_seller(catalogue)
-        self.with_context(product_company_id=company_id)._create_products(
-            catalogue, seller, filename=self.product_filename
+        wiz = self.with_context(product_company_id=company_id)
+        # Create products asynchronously
+        iterators = [iter(catalogue["products"])] * chunk_size
+        for products in zip_longest(*iterators):
+            if products[-1] is None:
+                products = [product for product in products if product]
+            # One job for x products (chunk of 40)
+            wiz.with_delay()._create_update_products(products, seller.id)
+        # Save imported file as attachment
+        self._bdimport.post_create_or_update(
+            catalogue, seller, doc_filename=self.product_filename
         )
+        logger.info(
+            "Update for vendor %s: %d products", seller.name, len(catalogue["products"])
+        )
+
         return {"type": "ir.actions.act_window_close"}

product_import_ubl/tests/test_ubl_catalogue_import.py

@@ -11,7 +11,13 @@ class TestUblOrderImport(SavepointCase):
     @classmethod
     def setUpClass(cls):
         super().setUpClass()
-        cls.env = cls.env(context=dict(cls.env.context, tracking_disable=True))
+        cls.env = cls.env(
+            context={
+                **cls.env.context,
+                "tracking_disable": True,
+                "queue_job__no_delay": True,
+            }
+        )
         cls.supplier = cls.env["res.partner"].create(
             {"name": "Medical", "ref": "78456123"}
         )