fix(clients): allow chunked requests on WithTransformation methods (generated)

algolia/api-clients-automation#5011

Co-authored-by: algolia-bot <[email protected]>
Co-authored-by: Clément Vannicatte <[email protected]>

Author: algolia-bot

algoliasearch/src/main/java/com/algolia/api/IngestionClient.java

@@ -10,6 +10,7 @@
 import com.algolia.model.ingestion.*;
 import com.algolia.utils.*;
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.EnumSet;
@@ -4845,4 +4846,103 @@ public CompletableFuture<WatchResponse> validateSourceBeforeUpdateAsync(@Nonnull
     throws AlgoliaRuntimeException {
     return this.validateSourceBeforeUpdateAsync(sourceID, sourceUpdate, null);
   }
+
+  private <T> List<PushTaskRecords> objectsToPushTaskRecords(Iterable<T> objects) {
+    try {
+      ObjectMapper mapper = new ObjectMapper();
+      String json = mapper.writeValueAsString(objects);
+
+      return mapper.readValue(json, new TypeReference<List<PushTaskRecords>>() {});
+    } catch (Exception e) {
+      throw new AlgoliaRuntimeException("each object must have an `objectID` key in order to be indexed");
+    }
+  }
+
+  /**
+   * 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/).
+   *
+   * @summary Helper: Chunks the given `objects` list in subset of 1000 elements max in order to
+   *     make it fit in `batch` requests.
+   * @param indexName - The `indexName` to replace `objects` in.
+   * @param objects - The array of `objects` to store in the given Algolia `indexName`.
+   * @param action - The `batch` `action` to perform on the given array of `objects`.
+   * @param waitForTasks - Whether or not we should wait until every `batch` tasks has been
+   *     processed, this operation may slow the total execution time of this method but is more
+   *     reliable.
+   * @param batchSize - The size of the chunk of `objects`. The number of `batch` calls will be
+   *     equal to `length(objects) / batchSize`. Defaults to 1000.
+   * @param referenceIndexName - This is required when targeting an index that does not have a push
+   *     connector setup (e.g. a tmp index), but you wish to attach another index's transformation
+   *     to it (e.g. the source index name).
+   * @param requestOptions - The requestOptions to send along with the query, they will be forwarded
+   *     to the `getEvent` method and merged with the transporter requestOptions.
+   */
+  public <T> List<WatchResponse> chunkedPush(
+    String indexName,
+    Iterable<T> objects,
+    Action action,
+    boolean waitForTasks,
+    int batchSize,
+    String referenceIndexName,
+    RequestOptions requestOptions
+  ) {
+    List<WatchResponse> responses = new ArrayList<>();
+    List<T> records = new ArrayList<>();
+
+    for (T item : objects) {
+      if (records.size() == batchSize) {
+        WatchResponse watch =
+          this.push(
+              indexName,
+              new PushTaskPayload().setAction(action).setRecords(this.objectsToPushTaskRecords(records)),
+              waitForTasks,
+              referenceIndexName,
+              requestOptions
+            );
+        responses.add(watch);
+        records.clear();
+      }
+
+      records.add(item);
+    }
+
+    if (records.size() > 0) {
+      WatchResponse watch =
+        this.push(
+            indexName,
+            new PushTaskPayload().setAction(action).setRecords(this.objectsToPushTaskRecords(records)),
+            waitForTasks,
+            referenceIndexName,
+            requestOptions
+          );
+      responses.add(watch);
+    }
+
+    if (waitForTasks) {
+      responses.forEach(response -> {
+        TaskUtils.retryUntil(
+          () -> {
+            try {
+              return this.getEvent(response.getRunID(), response.getEventID());
+            } catch (AlgoliaApiException e) {
+              if (e.getStatusCode() == 404) {
+                return null;
+              }
+
+              throw e;
+            }
+          },
+          (Event resp) -> {
+            return resp != null;
+          },
+          50,
+          null
+        );
+      });
+    }
+
+    return responses;
+  }
 }

algoliasearch/src/main/java/com/algolia/api/SearchClient.java

@@ -8,13 +8,10 @@
 import com.algolia.config.ClientOptions;
 import com.algolia.exceptions.*;
 import com.algolia.internal.JsonSerializer;
-import com.algolia.model.ingestion.PushTaskPayload;
-import com.algolia.model.ingestion.PushTaskRecords;
 import com.algolia.model.ingestion.WatchResponse;
 import com.algolia.model.search.*;
 import com.algolia.utils.*;
 import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import java.nio.charset.Charset;
 import java.security.InvalidKeyException;
 import java.security.NoSuchAlgorithmException;
@@ -6765,7 +6762,7 @@ public <T> List<BatchResponse> chunkedBatch(
    * @throws AlgoliaApiException When the API sends an http error code
    * @throws AlgoliaRuntimeException When an error occurred during the serialization
    */
-  public <T> WatchResponse saveObjectsWithTransformation(String indexName, Iterable<T> objects) {
+  public <T> List<WatchResponse> saveObjectsWithTransformation(String indexName, Iterable<T> objects) {
     return saveObjectsWithTransformation(indexName, objects, null);
   }
 
@@ -6780,7 +6777,7 @@ public <T> WatchResponse saveObjectsWithTransformation(String indexName, Iterabl
    * @param requestOptions The requestOptions to send along with the query, they will be merged with
    *     the transporter requestOptions. (optional)
    */
-  public <T> WatchResponse saveObjectsWithTransformation(String indexName, Iterable<T> objects, RequestOptions requestOptions) {
+  public <T> List<WatchResponse> saveObjectsWithTransformation(String indexName, Iterable<T> objects, RequestOptions requestOptions) {
     return saveObjectsWithTransformation(indexName, objects, false, requestOptions);
   }
 
@@ -6798,7 +6795,7 @@ public <T> WatchResponse saveObjectsWithTransformation(String indexName, Iterabl
    * @param requestOptions The requestOptions to send along with the query, they will be merged with
    *     the transporter requestOptions. (optional)
    */
-  public <T> WatchResponse saveObjectsWithTransformation(
+  public <T> List<WatchResponse> saveObjectsWithTransformation(
     String indexName,
     Iterable<T> objects,
     boolean waitForTasks,
@@ -6823,7 +6820,7 @@ public <T> WatchResponse saveObjectsWithTransformation(
    * @param requestOptions The requestOptions to send along with the query, they will be merged with
    *     the transporter requestOptions. (optional)
    */
-  public <T> WatchResponse saveObjectsWithTransformation(
+  public <T> List<WatchResponse> saveObjectsWithTransformation(
     String indexName,
     Iterable<T> objects,
     boolean waitForTasks,
@@ -6834,28 +6831,17 @@ public <T> WatchResponse saveObjectsWithTransformation(
       throw new AlgoliaRuntimeException("`setTransformationRegion` must have been called before calling this method.");
     }
 
-    return this.ingestionTransporter.push(
+    return this.ingestionTransporter.chunkedPush(
         indexName,
-        new PushTaskPayload().setAction(com.algolia.model.ingestion.Action.ADD_OBJECT).setRecords(this.objectsToPushTaskRecords(objects)),
+        objects,
+        com.algolia.model.ingestion.Action.ADD_OBJECT,
         waitForTasks,
+        batchSize,
         null,
         requestOptions
       );
   }
 
-  private <T> List<PushTaskRecords> objectsToPushTaskRecords(Iterable<T> objects) {
-    try {
-      ObjectMapper mapper = new ObjectMapper();
-      String json = mapper.writeValueAsString(objects);
-
-      return mapper.readValue(json, new TypeReference<List<PushTaskRecords>>() {});
-    } catch (Exception e) {
-      throw new AlgoliaRuntimeException(
-        "each object must have an `objectID` key in order to be used with the" + " WithTransformation methods"
-      );
-    }
-  }
-
   /**
    * Helper: Saves the given array of objects in the given index. The `chunkedBatch` helper is used
    * under the hood, which creates a `batch` requests with at most 1000 objects in it.
@@ -7003,7 +6989,7 @@ public List<BatchResponse> deleteObjects(
    * @param createIfNotExists To be provided if non-existing objects are passed, otherwise, the call
    *     will fail.
    */
-  public <T> WatchResponse partialUpdateObjectsWithTransformation(String indexName, Iterable<T> objects, boolean createIfNotExists) {
+  public <T> List<WatchResponse> partialUpdateObjectsWithTransformation(String indexName, Iterable<T> objects, boolean createIfNotExists) {
     return partialUpdateObjectsWithTransformation(indexName, objects, createIfNotExists, false, null);
   }
 
@@ -7021,7 +7007,7 @@ public <T> WatchResponse partialUpdateObjectsWithTransformation(String indexName
    *     processed, this operation may slow the total execution time of this method but is more
    *     reliable.
    */
-  public <T> WatchResponse partialUpdateObjectsWithTransformation(
+  public <T> List<WatchResponse> partialUpdateObjectsWithTransformation(
     String indexName,
     Iterable<T> objects,
     boolean createIfNotExists,
@@ -7046,7 +7032,7 @@ public <T> WatchResponse partialUpdateObjectsWithTransformation(
    * @param requestOptions The requestOptions to send along with the query, they will be merged with
    *     the transporter requestOptions. (optional)
    */
-  public <T> WatchResponse partialUpdateObjectsWithTransformation(
+  public <T> List<WatchResponse> partialUpdateObjectsWithTransformation(
     String indexName,
     Iterable<T> objects,
     boolean createIfNotExists,
@@ -7074,7 +7060,7 @@ public <T> WatchResponse partialUpdateObjectsWithTransformation(
    * @param requestOptions The requestOptions to send along with the query, they will be merged with
    *     the transporter requestOptions. (optional)
    */
-  public <T> WatchResponse partialUpdateObjectsWithTransformation(
+  public <T> List<WatchResponse> partialUpdateObjectsWithTransformation(
     String indexName,
     Iterable<T> objects,
     boolean createIfNotExists,
@@ -7086,16 +7072,14 @@ public <T> WatchResponse partialUpdateObjectsWithTransformation(
       throw new AlgoliaRuntimeException("`setTransformationRegion` must have been called before calling this method.");
     }
 
-    return this.ingestionTransporter.push(
+    return this.ingestionTransporter.chunkedPush(
         indexName,
-        new PushTaskPayload()
-          .setAction(
-            createIfNotExists
-              ? com.algolia.model.ingestion.Action.PARTIAL_UPDATE_OBJECT
-              : com.algolia.model.ingestion.Action.PARTIAL_UPDATE_OBJECT_NO_CREATE
-          )
-          .setRecords(this.objectsToPushTaskRecords(objects)),
+        objects,
+        createIfNotExists
+          ? com.algolia.model.ingestion.Action.PARTIAL_UPDATE_OBJECT
+          : com.algolia.model.ingestion.Action.PARTIAL_UPDATE_OBJECT_NO_CREATE,
         waitForTasks,
+        batchSize,
         null,
         requestOptions
       );