fix: py

Author: shortcuts

templates/python/ingestion_helpers.mustache

@@ -11,6 +11,7 @@
         """
         Helper: Chunks the given `objects` list in subset of 1000 elements max in order to make it fit in `push` requests by leveraging the Transformation pipeline setup in the Push connector (https://www.algolia.com/doc/guides/sending-and-managing-data/send-and-update-your-data/connectors/push/).
         """
+        offset = 0
         records: List[PushTaskRecords] = []
         responses: List[WatchResponse] = []
         wait_batch_size = batch_size // 10
@@ -20,7 +21,7 @@
             records.append(obj)  # pyright: ignore
             if len(records) == batch_size or i == len(objects) - 1:
                 responses.append(
-                    {{^isSyncClient}}await {{/isSyncClient}}self.push(
+                    await self.push(
                         index_name=index_name,
                         push_task_payload={
                             "action": action,
@@ -31,18 +32,27 @@
                     )
                 )
                 records = []
-            if wait_for_tasks and (len(responses) % wait_batch_size == 0 or i == len(objects) - 1):
-                for response in responses[-wait_batch_size:]:
-                    {{^isSyncClient}}async {{/isSyncClient}}def _func(_: Optional[Event]) -> Event:
+            if (
+                wait_for_tasks
+                and len(responses) > 0
+                and (len(responses) % wait_batch_size == 0 or i == len(objects) - 1)
+            ):
+                for response in responses[offset:offset+wait_batch_size]:
+
+                    async def _func(_: Optional[Event]) -> Event:
                         if response.event_id is None:
                             raise ValueError(
                                 "received unexpected response from the push endpoint, eventID must not be undefined"
                             )
                         try:
-                            return {{^isSyncClient}}await {{/isSyncClient}}self.get_event(run_id=response.run_id, event_id=response.event_id, request_options=request_options)
+                            return await self.get_event(
+                                run_id=response.run_id,
+                                event_id=response.event_id,
+                                request_options=request_options,
+                            )
                         except RequestException as e:
                             if e.status_code == 404:
-                                return None # pyright: ignore
+                                return None  # pyright: ignore
                             raise e
 
                     _retry_count = 0
@@ -56,12 +66,13 @@
 
                     timeout = RetryTimeout()
 
-                    {{^isSyncClient}}await {{/isSyncClient}}create_iterable{{#isSyncClient}}_sync{{/isSyncClient}}(
+                    await create_iterable(
                         func=_func,
                         validate=_validate,
                         aggregator=_aggregator,
                         timeout=lambda: timeout(_retry_count),
                         error_validate=lambda _: _retry_count >= 50,
                         error_message=lambda _: f"The maximum number of retries exceeded. (${_retry_count}/${50})",
                     )
+                offset += wait_batch_size
         return responses