refactor: enhance stock sending logic to include retry mechanism and handle rate limits

marketplaces/ozon.py

@@ -47,32 +47,35 @@ class OzonMarketplaceApi(BaseMarketplaceApi):
 
         self.init_session()
         limiter = BatchLimiter()
+        max_retries = 5
+        while chunks:
+            current_retry = 0
+            chunk = chunks.pop()
+            while current_retry <= max_retries:
+                try:
+                    await limiter.acquire_ozon(self.limiter_key)
+                    request_data = {'stocks': chunk}
+                    response = await self._method('POST', '/v2/products/stocks', data=request_data)
+                    current_retry += 1
+                    response = await response.json()
+                    error_message = response.get('message')
+                    error_code = response.get('code')
+                    if error_message:
+                        if error_code == 8:
+                            logging.warning(f'Ozon rate limit exceeded for marketplace [{self.marketplace.id}]')
+                            await asyncio.sleep(1)
+                            continue
+                        else:
+                            logging.warning(
+                                f'Error occurred when sending stocks to [{self.marketplace.id}]: {error_message} ({error_code})')
+                            break
+                    else:
+                        break
 
-        async def send_stock_chunk(chunk) -> bool:
+                except Exception as e:
-            try:
+                    logging.error(
-                await limiter.acquire_ozon(self.limiter_key)
+                        f'Exception occurred while sending stocks to marketplace ID [{self.marketplace.id}]: {str(e)}')
-                request_data = {'stocks': chunk}
+                    break
-                response = await self._method('POST', '/v2/products/stocks', data=request_data)
-                response = await response.json()
-                error_message = response.get('message')
-                error_code = response.get('code')
-                if error_message:
-                    logging.warning(
-                        f'Error occurred when sending stocks to [{self.marketplace.id}]: {error_message} ({error_code})')
-                    return False
-                return True
-            except Exception as e:
-                logging.error(
-                    f'Exception occurred while sending stocks to marketplace ID [{self.marketplace.id}]: {str(e)}')
-                return False
 
-        tasks = [send_stock_chunk(chunk) for chunk in chunks]
-        first_request = tasks[0]
-        first_response = await first_request
-        if not first_response:
-            logging.error(f'Skipping marketplace [{self.marketplace.id}] because first request was unsuccessful')
-            await self.session.close()
-            return
 
-        await asyncio.gather(*tasks[1:])
         await self.session.close()

marketplaces/wildberries.py

@@ -50,40 +50,32 @@ class WildberriesMarketplaceApi(BaseMarketplaceApi):
             return
         self.init_session()
         limiter = BatchLimiter()
-
+        max_retries = 5
-        chunk = chunks.pop()
+        while chunks:
-        while True:
+            current_retry = 0
-            try:
+            chunk = chunks.pop()
-                if not chunks:
+            while current_retry <= max_retries:
+                try:
+                    await limiter.acquire_wildberries(self.limiter_key)
+                    request_data = {'stocks': chunk}
+                    response = await self._method('PUT', f'/api/v3/stocks/{self.marketplace.warehouse_id}',
+                                                  data=request_data)
+                    current_retry += 1
+                    if (response.status not in [204, 409]) and (response.status != 429):
+                        response = await response.json()
+                        error_message = response.get('message')
+                        error_code = response.get('code')
+                        logging.warning(
+                            f'Error occurred when sending stocks to [{self.marketplace.id}]: {error_message} ({error_code})')
+                        break
+                    if response.status == 429:
+                        logging.warning(f'WB rate limit exceeded for marketplace [{self.marketplace.id}]')
+                        await asyncio.sleep(1)
+                        continue
+                    await asyncio.sleep(0.2)
                     break
-                await limiter.acquire_wildberries(self.limiter_key)
+                except Exception as e:
-                request_data = {'stocks': chunk}
+                    logging.error(
-                response = await self._method('PUT', f'/api/v3/stocks/{self.marketplace.warehouse_id}',
+                        f'Exception occurred while sending stocks to marketplace ID [{self.marketplace.id}]: {str(e)}')
-                                              data=request_data)
-                if (response.status not in [204, 409]) and (response.status != 429):
-                    response = await response.json()
-                    error_message = response.get('message')
-                    error_code = response.get('code')
-                    logging.warning(
-                        f'Error occurred when sending stocks to [{self.marketplace.id}]: {error_message} ({error_code})')
                     break
-                if response.status == 429:
-                    await asyncio.sleep(1)
-                    logging.warning(f'Rate limit exceeded for marketplace [{self.marketplace.id}]')
-                    continue
-                chunk = chunks.pop()
-                await asyncio.sleep(0.2)
-
-            except Exception as e:
-                logging.error(
-                    f'Exception occurred while sending stocks to marketplace ID [{self.marketplace.id}]: {str(e)}')
-        # tasks = [send_stock_chunk(chunk) for chunk in chunks]
-        # first_request = tasks[0]
-        # first_response = await first_request
-        # if not first_response:
-        #     logging.error(f'Skipping marketplace [{self.marketplace.id}] because first request was unsuccessful')
-        #     await self.session.close()
-        #     return
-        #
-        # await asyncio.gather(*tasks[1:])
         await self.session.close()