feat: python

Author: shortcuts

templates/python/api.mustache

@@ -12,7 +12,7 @@ from algoliasearch.search.models import (
   SecuredApiKeyRestrictions,
 )
 
-from algoliasearch.ingestion.models import WatchResponse
+from algoliasearch.ingestion.models import (WatchResponse, Event, PushTaskRecords)
 from algoliasearch.ingestion.config import IngestionConfig
 from algoliasearch.ingestion.client import (IngestionClient, IngestionClientSync)
 {{/isSearchClient}}

templates/python/search_helpers.mustache

@@ -74,7 +74,7 @@
                 "`apiKey` is required when waiting for an `update` operation."
             )
 
-        {{^isSyncClient}}async {{/isSyncClient}}def _func(_prev: Optional[GetApiKeyResponse]) -> GetApiKeyResponse:
+        {{^isSyncClient}}async {{/isSyncClient}}def _func(_: Optional[GetApiKeyResponse]) -> GetApiKeyResponse:
             try:
                 return {{^isSyncClient}}await {{/isSyncClient}}self.get_api_key(key=key, request_options=request_options)
             except RequestException as e:
@@ -209,7 +209,7 @@
         page = search_synonyms_params.page or 0
         search_synonyms_params.hits_per_page = hits_per_page
 
-        {{^isSyncClient}}async {{/isSyncClient}}def _func(_prev: Optional[SearchSynonymsResponse]) -> SearchSynonymsResponse:
+        {{^isSyncClient}}async {{/isSyncClient}}def _func(_: Optional[SearchSynonymsResponse]) -> SearchSynonymsResponse:
             nonlocal page
             resp = {{^isSyncClient}}await {{/isSyncClient}}self.search_synonyms(
                 index_name=index_name,
@@ -303,19 +303,7 @@
         """
         Helper: Similar to the `save_objects` method but requires a Push connector (https://www.algolia.com/doc/guides/sending-and-managing-data/send-and-update-your-data/connectors/push/) to be created first, in order to transform records before indexing them to Algolia. The `region` must've been passed to the client's config at instantiation.
         """
-        if self._ingestion_transporter is None:
-            raise ValueError("`region` must be provided at client instantiation before calling this method.")
-
-        return {{^isSyncClient}}await {{/isSyncClient}}self._ingestion_transporter.push(
-            index_name=index_name,
-            push_task_payload={
-                "action": Action.ADDOBJECT,
-                "records": objects,
-            },
-            watch=wait_for_tasks,
-            request_options=request_options,
-        )
-
+        return {{^isSyncClient}}await {{/isSyncClient}}self.chunked_push(index_name=index_name, objects=objects, action=Action.ADDOBJECT, wait_for_tasks=wait_for_tasks, batch_size=batch_size, request_options=request_options)
 
     {{^isSyncClient}}async {{/isSyncClient}}def delete_objects(
         self,
@@ -356,19 +344,80 @@
         """
         Helper: Similar to the `partial_update_objects` method but requires a Push connector (https://www.algolia.com/doc/guides/sending-and-managing-data/send-and-update-your-data/connectors/push/) to be created first, in order to transform records before indexing them to Algolia. The `region` must've been passed to the client instantiation method.
         """
-        if self._ingestion_transporter is None:
-            raise ValueError("`region` must be provided at client instantiation before calling this method.")
+        return {{^isSyncClient}}await {{/isSyncClient}}self.chunked_push(index_name=index_name, objects=objects, action=Action.PARTIALUPDATEOBJECT if create_if_not_exists else Action.PARTIALUPDATEOBJECTNOCREATE, wait_for_tasks=wait_for_tasks, batch_size=batch_size, request_options=request_options)
 
-        return {{^isSyncClient}}await {{/isSyncClient}}self._ingestion_transporter.push(
-            index_name=index_name,
-            push_task_payload={
-                "action": Action.PARTIALUPDATEOBJECT if create_if_not_exists else Action.PARTIALUPDATEOBJECTNOCREATE,
-                "records": objects,
-            },
-            watch=wait_for_tasks,
-            request_options=request_options,
-        )
+    {{^isSyncClient}}async {{/isSyncClient}}def chunked_push(
+        self,
+        index_name: str,
+        objects: List[Dict[str, Any]],
+        action: Action = Action.ADDOBJECT,
+        wait_for_tasks: bool = False,
+        batch_size: int = 1000,
+        reference_index_name: Optional[str] = None,
+        request_options: Optional[Union[dict, RequestOptions]] = None,
+    ) -> List[WatchResponse]:
+        """
+        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/).
+        """
 
+        if self._ingestion_transporter is None:
+            raise ValueError("`region` must be provided at client instantiation before calling this method.")
+        records: List[PushTaskRecords] = []
+        responses: List[WatchResponse] = []
+        for i, obj in enumerate(objects):
+            records.append(obj)  # pyright: ignore
+            if len(records) == batch_size or i == len(objects) - 1:
+                responses.append(
+                    {{^isSyncClient}}await {{/isSyncClient}}self._ingestion_transporter.push(
+                        index_name=index_name,
+                        push_task_payload={
+                            "action": action,
+                            "records": records,
+                        },
+                        watch=wait_for_tasks,
+                        reference_index_name=reference_index_name,
+                        request_options=request_options,
+                    )
+                )
+                requests = []
+        if wait_for_tasks:
+            for response in responses:
+                {{^isSyncClient}}async {{/isSyncClient}}def _func(_: Optional[Event]) -> Event:
+                    if self._ingestion_transporter is None:
+                        raise ValueError(
+                            "`region` must be provided at client instantiation before calling this method."
+                        )
+                    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._ingestion_transporter.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
+                        raise e
+
+                _retry_count = 0
+
+                def _aggregator(_: Event | None) -> None:
+                    nonlocal _retry_count
+                    _retry_count += 1
+
+                def _validate(_resp: Event | None) -> bool:
+                    return _resp is not None
+
+                timeout = RetryTimeout()
+
+                {{^isSyncClient}}await {{/isSyncClient}}create_iterable{{#isSyncClient}}_sync{{/isSyncClient}}(
+                    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})",
+                )
+        return responses
 
     {{^isSyncClient}}async {{/isSyncClient}}def chunked_batch(
         self,