feat(python): add bridge to transformation on search (#4898)

Author: shortcuts

playground/python/app/search.py

@@ -1,6 +1,7 @@
 from os import environ
 
 from algoliasearch.search.client import SearchClientSync
+from algoliasearch.search.config import SearchConfig
 from algoliasearch.search import __version__
 from dotenv import load_dotenv
 
@@ -27,3 +28,29 @@ def main():
         client.close()
 
         print("client closed")
+
+    print("with transformations")
+
+    config = SearchConfig(
+        environ.get("ALGOLIA_APPLICATION_ID"), environ.get("ALGOLIA_ADMIN_KEY")
+    )
+
+    config.with_transformation("eu")
+
+    print("region in playground")
+    print(config.region)
+
+    client = SearchClientSync.create_with_config(config)
+    client.add_user_agent("playground search with ingestion")
+
+    print("user_agent", client._config._user_agent.get())
+
+    try:
+        resp = client.save_objects_with_transformation(
+            "foo", [{"objectID": "bar"}], wait_for_tasks=True
+        )
+        print(resp)
+    finally:
+        client.close()
+
+        print("client closed")

playground/python/poetry.lock

@@ -0,0 +1,6 @@
+argcomplete==3.6.2
+click==8.2.1
+packaging==25.0
+pipx==1.7.1
+platformdirs==4.3.8
+userpath==1.9.2

playground/python/requirements.txt

@@ -9,6 +9,10 @@ from algoliasearch.search.models.operation_type import OperationType
 from algoliasearch.search.models.replace_all_objects_response import ReplaceAllObjectsResponse
 from algoliasearch.search.models.scope_type import ScopeType
 from algoliasearch.search.models.secured_api_key_restrictions import SecuredApiKeyRestrictions
+
+from algoliasearch.ingestion.models.watch_response import WatchResponse
+from algoliasearch.ingestion.config import IngestionConfig
+from algoliasearch.ingestion.client import (IngestionClient, IngestionClientSync)
 {{/isSearchClient}}
 
 {{#operations}}{{#operation}}{{#imports}}
@@ -37,15 +41,22 @@ class {{classname}}{{#isSyncClient}}Sync{{/isSyncClient}}:
     """
 
     _transporter: Transporter{{#isSyncClient}}Sync{{/isSyncClient}}
+    {{#isSearchClient}}
+    _ingestion_transporter: Optional[IngestionClient{{#isSyncClient}}Sync{{/isSyncClient}}]
+    {{/isSearchClient}}
     _config: BaseConfig
     _request_options: RequestOptions
 
     def __init__(self, app_id: Optional[str] = None, api_key: Optional[str] = None, {{#hasRegionalHost}}region: {{#fallbackToAliasHost}}Optional[str] = None{{/fallbackToAliasHost}}{{^fallbackToAliasHost}}str = ""{{/fallbackToAliasHost}}, {{/hasRegionalHost}}transporter: Optional[Transporter{{#isSyncClient}}Sync{{/isSyncClient}}] = None, config: Optional[{{#lambda.pascalcase}}{{client}}Config{{/lambda.pascalcase}}] = None) -> None:
         if transporter is not None and config is None:
             config = {{#lambda.pascalcase}}{{client}}Config{{/lambda.pascalcase}}(transporter.config.app_id, transporter.config.api_key{{#hasRegionalHost}}, region{{/hasRegionalHost}})
-
-        if config is None:
+        elif config is None:
             config = {{#lambda.pascalcase}}{{client}}Config{{/lambda.pascalcase}}(app_id, api_key{{#hasRegionalHost}}, region{{/hasRegionalHost}})
+        {{#isSearchClient}}
+        elif config.region is not None:
+            self._ingestion_transporter = IngestionClient{{#isSyncClient}}Sync{{/isSyncClient}}.create_with_config(IngestionConfig(config.app_id, config.api_key, config.region))
+        {{/isSearchClient}}
+
         self._config = config
         self._request_options = RequestOptions(config)
 

templates/python/api.mustache

@@ -77,3 +77,11 @@ class {{#lambda.pascalcase}}{{client}}{{/lambda.pascalcase}}Config(BaseConfig):
             {{/hostWithAppID}}
         )
         {{/hasRegionalHost}}
+
+    {{#isSearchClient}}
+        self.region = None
+
+    def with_transformation(self, region: str = ""):
+        "This method is required to be called with the appropriate region of your Algolia application if you wish to leverage the *_with_transformation methods."
+        self.region = region
+    {{/isSearchClient}}

templates/python/config.mustache

@@ -292,6 +292,31 @@
         """
         return {{^isSyncClient}}await {{/isSyncClient}}self.chunked_batch(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 save_objects_with_transformation(
+        self,
+        index_name: str,
+        objects: List[Dict[str, Any]],
+        wait_for_tasks: bool = False,
+        batch_size: int = 1000,
+        request_options: Optional[Union[dict, RequestOptions]] = None,
+    ) -> WatchResponse:
+        """
+        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,
+        )
+
+
     {{^isSyncClient}}async {{/isSyncClient}}def delete_objects(
         self,
         index_name: str,
@@ -319,6 +344,32 @@
         """
         return {{^isSyncClient}}await {{/isSyncClient}}self.chunked_batch(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 partial_update_objects_with_transformation(
+        self,
+        index_name: str,
+        objects: List[Dict[str, Any]],
+        create_if_not_exists: bool = False,
+        wait_for_tasks: bool = False,
+        batch_size: int = 1000,
+        request_options: Optional[Union[dict, RequestOptions]] = None,
+    ) -> WatchResponse:
+        """
+        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._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_batch(
         self,
         index_name: str,