feat: python

Author: shortcuts

templates/python/api.mustache

@@ -12,10 +12,13 @@ from algoliasearch.search.models import (
   SecuredApiKeyRestrictions,
 )
 
-from algoliasearch.ingestion.models import (WatchResponse, Event, PushTaskRecords)
+from algoliasearch.ingestion.models import (WatchResponse)
 from algoliasearch.ingestion.config import IngestionConfig
 from algoliasearch.ingestion.client import (IngestionClient, IngestionClientSync)
 {{/isSearchClient}}
+{{#isIngestionClient}}
+from algoliasearch.ingestion.models import (Action, WatchResponse, Event, PushTaskRecords)
+{{/isIngestionClient}}
 
 {{#operations}}{{#operation}}{{#imports}}
 from algoliasearch.{{packageName}}.models import {{{.}}}
@@ -130,6 +133,9 @@ class {{classname}}{{#isSyncClient}}Sync{{/isSyncClient}}:
     {{#isSearchClient}}
 {{> search_helpers}}
     {{/isSearchClient}}
+    {{#isIngestionClient}}
+{{> ingestion_helpers}}
+    {{/isIngestionClient}}
 
     {{#operation}}
 

templates/python/ingestion_helpers.mustache

@@ -0,0 +1,64 @@
+    {{^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/).
+        """
+        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.push(
+                        index_name=index_name,
+                        push_task_payload={
+                            "action": action,
+                            "records": records,
+                        },
+                        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 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)
+                    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

templates/python/search_helpers.mustache

@@ -303,7 +303,9 @@
         """
         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.
         """
-        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)
+        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.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,
@@ -344,79 +346,9 @@
         """
         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.
         """
-        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)
-
-    {{^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,
-                        },
-                        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
+        return {{^isSyncClient}}await {{/isSyncClient}}self._ingestion_transporter.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)
 
     {{^isSyncClient}}async {{/isSyncClient}}def chunked_batch(
         self,