solving conflicts 2

marketplace/clients/decorators.py

@@ -0,0 +1,28 @@
+import time
+import functools
+
+
+def retry_on_exception(max_attempts=11, start_sleep_time=1, factor=2):
+    def decorator_retry(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            attempts, sleep_time = 0, start_sleep_time
+            while attempts < max_attempts:
+                try:
+                    return func(*args, **kwargs)
+                except (
+                    Exception
+                ) as e:  # TODO: Map only timeout errors or errors from many requests
+                    print(
+                        f"Retrying... Attempt {attempts + 1} after {sleep_time} seconds, {str(e)}"
+                    )
+                    time.sleep(sleep_time)
+                    attempts += 1
+                    sleep_time *= factor
+
+            print("Max retry attempts reached. Raising exception.")
+            raise Exception("Rate limit exceeded, max retry attempts reached.")
+
+        return wrapper
+
+    return decorator_retry

marketplace/clients/flows/client.py

@@ -2,6 +2,7 @@
 from django.conf import settings
 
 from marketplace.clients.base import RequestClient
+from marketplace.clients.decorators import retry_on_exception
 
 
 class InternalAuthentication(RequestClient):
@@ -102,6 +103,8 @@ def update_facebook_templates(self, flow_object_uuid, fba_templates):
         )
         return response
 
+
+    @retry_on_exception()
     def update_vtex_products(self, products, flow_object_uuid, dict_catalog):
         data = {
             "catalog": dict_catalog,

marketplace/clients/vtex/client.py

@@ -1,4 +1,5 @@
 from marketplace.clients.base import RequestClient
+from marketplace.clients.decorators import retry_on_exception
 
 
 class VtexAuthorization(RequestClient):
@@ -45,6 +46,7 @@ def is_valid_credentials(self, domain):
         except Exception:
             return False
 
+    @retry_on_exception()
     def list_all_products_sku_ids(self, domain, page_size=1000):
         all_skus = []
         page = 1
@@ -63,13 +65,15 @@ def list_all_products_sku_ids(self, domain, page_size=1000):
 
         return all_skus
 
+    @retry_on_exception()
     def list_active_sellers(self, domain):
         url = f"https://{domain}/api/seller-register/pvt/sellers"
         headers = self._get_headers()
         response = self.make_request(url, method="GET", headers=headers)
         sellers_data = response.json()
         return [seller["id"] for seller in sellers_data["items"] if seller["isActive"]]
 
+    @retry_on_exception()
     def get_product_details(self, sku_id, domain):
         url = (
             f"https://{domain}/api/catalog_system/pvt/sku/stockkeepingunitbyid/{sku_id}"
@@ -78,6 +82,7 @@ def get_product_details(self, sku_id, domain):
         response = self.make_request(url, method="GET", headers=headers)
         return response.json()
 
+    @retry_on_exception()
     def pub_simulate_cart_for_seller(self, sku_id, seller_id, domain):
         cart_simulation_url = f"https://{domain}/api/checkout/pub/orderForms/simulation"
         payload = {"items": [{"id": sku_id, "quantity": 1, "seller": seller_id}]}

marketplace/services/product/product_facebook_manage.py

@@ -1,6 +1,5 @@
 from typing import List
 
-from django.db import transaction
 from django.contrib.auth import get_user_model
 
 from marketplace.services.vtex.utils.data_processor import FacebookProductDTO
@@ -11,34 +10,7 @@
 
 
 class ProductFacebookManager:
-    def save_products_on_database(
-        self, products: List[FacebookProductDTO], catalog, product_feed
-    ):
-        product_instances = [
-            Product(
-                facebook_product_id=dto.id,
-                title=dto.title,
-                description=dto.description,
-                availability=dto.availability,
-                condition=dto.condition,
-                price=dto.price,
-                link=dto.link,
-                image_link=dto.image_link,
-                brand=dto.brand,
-                sale_price=dto.sale_price,
-                catalog=catalog,
-                created_by=catalog.created_by,
-                feed=product_feed,
-            )
-            for dto in products
-        ]
-
-        with transaction.atomic():
-            Product.objects.bulk_create(product_instances)
-
-        return True
-
-    def update_products_on_database(
+    def create_or_update_products_on_database(
         self, products: List[FacebookProductDTO], catalog, product_feed
     ):
         products_to_update = []
@@ -47,7 +19,7 @@ def update_products_on_database(
         for dto in products:
             try:
                 product = Product.objects.get(
-                    facebook_product_id=dto.id, catalog=catalog, feed=product_feed
+                    facebook_product_id=dto.id, catalog=catalog
                 )  # TODO: Optimize to make a single query at the bank
                 product.title = dto.title
                 product.description = dto.description

marketplace/services/vtex/generic_service.py

@@ -79,9 +79,9 @@ def configure(self, app, credentials: APICredentials, wpp_cloud_uuid) -> App:
         app.config["wpp_cloud_uuid"] = wpp_cloud_uuid
         app.config["initial_sync_completed"] = False
         app.config["rules"] = [
+            "exclude_alcoholic_drinks",
             "calculate_by_weight",
             "currency_pt_br",
-            "exclude_alcoholic_drinks",
             "unifies_id_with_seller",
         ]
         app.configured = True
@@ -106,7 +106,9 @@ def first_product_insert(self, credentials: APICredentials, catalog: Catalog):
         )
         products_csv = pvt_service.data_processor.products_to_csv(products)
         product_feed = self._send_products_to_facebook(products_csv, catalog)
-        self.product_manager.save_products_on_database(products, catalog, product_feed)
+        self.product_manager.create_or_update_products_on_database(
+            products, catalog, product_feed
+        )
         self.app_manager.initial_sync_products_completed(catalog.vtex_app)
 
         return pvt_service.data_processor.convert_dtos_to_dicts_list(products)
@@ -125,7 +127,7 @@ def webhook_product_insert(
 
         products_csv = pvt_service.data_processor.products_to_csv(products_dto)
         self._update_products_on_facebook(products_csv, catalog, product_feed)
-        self.product_manager.update_products_on_database(
+        self.product_manager.create_or_update_products_on_database(
             products_dto, catalog, product_feed
         )
         return pvt_service.data_processor.convert_dtos_to_dicts_list(products_dto)

marketplace/services/vtex/utils/data_processor.py

@@ -1,6 +1,9 @@
+import concurrent.futures
 import pandas as pd
 import io
 import dataclasses
+import os
+import time
 
 from dataclasses import dataclass
 
@@ -69,27 +72,60 @@ def extract_fields(product_details, availability_details) -> FacebookProductDTO:
 
     @staticmethod
     def process_product_data(
-        skus_ids, active_sellers, service, domain, rules, update_product=False
+        skus_ids,
+        active_sellers,
+        service,
+        domain,
+        rules,
+        update_product=False,
     ):
-        facebook_products = []
-        for sku_id in skus_ids:
-            product_details = service.get_product_details(sku_id, domain)
-            for seller_id in active_sellers:
-                availability_details = service.simulate_cart_for_seller(
-                    sku_id, seller_id, domain
-                )
-                if update_product is False and not availability_details["is_available"]:
-                    continue
+        num_cpus = os.cpu_count()
+        all_facebook_products = []
 
-                product_dto = DataProcessor.extract_fields(
-                    product_details, availability_details
+        with concurrent.futures.ThreadPoolExecutor(max_workers=num_cpus) as executor:
+            futures = [
+                executor.submit(
+                    DataProcessor.process_single_sku,
+                    sku_id,
+                    active_sellers,
+                    service,
+                    domain,
+                    rules,
+                    update_product,
                 )
-                params = {"seller_id": seller_id}
-                for rule in rules:
-                    if not rule.apply(product_dto, **params):
-                        break
-                else:
-                    facebook_products.append(product_dto)
+                for sku_id in skus_ids
+            ]
+            time.sleep(
+                15
+            )  # TODO: Test whether by waiting this time the processes are terminated correctly
+            for future in concurrent.futures.as_completed(futures):
+                try:
+                    all_facebook_products.extend(future.result())
+                    print("total products extend to", len(all_facebook_products))
+                except Exception as e:
+                    print(f"Exception in thread: {e}")
+
+        return all_facebook_products
+
+    @staticmethod
+    def process_single_sku(
+        sku_id, active_sellers, service, domain, rules, update_product
+    ):
+        facebook_products = []
+        product_details = service.get_product_details(sku_id, domain)
+        for seller_id in active_sellers:
+            availability_details = service.simulate_cart_for_seller(
+                sku_id, seller_id, domain
+            )
+            if update_product is False and not availability_details["is_available"]:
+                continue
+
+            product_dto = DataProcessor.extract_fields(
+                product_details, availability_details
+            )
+            params = {"seller_id": seller_id}
+            if all(rule.apply(product_dto, **params) for rule in rules):
+                facebook_products.append(product_dto)
 
         return facebook_products
 

marketplace/wpp_products/tasks.py

@@ -162,9 +162,9 @@ def task_update_vtex_products(**kwargs):
 
     app_uuid = kwargs.get("app_uuid")
     webhook_data = kwargs.get("webhook_data")
-    vtex_app = App.objects.get(uuid=app_uuid, configured=True, code="vtex")
 
     try:
+        vtex_app = App.objects.get(uuid=app_uuid, configured=True, code="vtex")
         domain, app_key, app_token = vtex_service.get_vtex_credentials_or_raise(
             vtex_app
         )