Merge #261

261: [Feature] [add/update]DocumentsInBatches r=brunoocasali a=ahmednfwela

Fixes #95
- [x] Tests added 

Co-authored-by: Ahmed Fwela <[email protected]>

Author: bors[bot]

lib/src/index.dart

@@ -62,13 +62,29 @@ abstract class MeiliSearchIndex {
     String? primaryKey,
   });
 
+  /// Add a list of documents in batches of size [batchSize] by given [documents] and optional [primaryKey] parameter.
+  /// If the index does not exist try to create a new index and add documents.
+  Future<List<Task>> addDocumentsInBatches(
+    List<Map<String, Object?>> documents, {
+    int batchSize = 1000,
+    String? primaryKey,
+  });
+
   /// Add a list of documents or update them if they already exist by given [documents] and optional [primaryKey] parameter.
   /// If index is not exists tries to create a new index and adds documents.
   Future<Task> updateDocuments(
     List<Map<String, Object?>> documents, {
     String? primaryKey,
   });
 
+  /// Add a list of documents or update them if they already exist in batches of size [batchSize] by given [documents] and optional [primaryKey] parameter.
+  /// If index is not exists tries to create a new index and adds documents.
+  Future<List<Task>> updateDocumentsInBatches(
+    List<Map<String, Object?>> documents, {
+    int batchSize = 1000,
+    String? primaryKey,
+  });
+
   /// Delete one document by given [id].
   Future<Task> deleteDocument(Object id);
 

lib/src/index_impl.dart

@@ -4,7 +4,7 @@ import 'package:meilisearch/src/query_parameters/tasks_query.dart';
 import 'package:meilisearch/src/result.dart';
 import 'package:meilisearch/src/searchable.dart';
 import 'package:meilisearch/src/tasks_results.dart';
-
+import 'package:collection/collection.dart';
 import 'client.dart';
 import 'filter_builder/filter_builder_base.dart';
 import 'index.dart';
@@ -448,4 +448,28 @@ class MeiliSearchIndexImpl implements MeiliSearchIndex {
   Future<Task> getTask(int uid) async {
     return await client.getTask(uid);
   }
+
+  @override
+  Future<List<Task>> addDocumentsInBatches(
+    List<Map<String, Object?>> documents, {
+    int batchSize = 1000,
+    String? primaryKey,
+  }) =>
+      Future.wait(
+        documents
+            .slices(batchSize)
+            .map((slice) => addDocuments(slice, primaryKey: primaryKey)),
+      );
+
+  @override
+  Future<List<Task>> updateDocumentsInBatches(
+    List<Map<String, Object?>> documents, {
+    int batchSize = 1000,
+    String? primaryKey,
+  }) =>
+      Future.wait(
+        documents
+            .slices(batchSize)
+            .map((slice) => updateDocuments(slice, primaryKey: primaryKey)),
+      );
 }

test/documents_test.dart

@@ -1,6 +1,5 @@
 import 'package:meilisearch/meilisearch.dart';
 import 'package:test/test.dart';
-
 import 'utils/books_data.dart';
 import 'utils/wait_for.dart';
 import 'utils/client.dart';
@@ -17,6 +16,23 @@ void main() {
       expect(docs.total, books.length);
     });
 
+    test('Add documents in batches', () async {
+      final index = client.index(randomUid());
+      const batchSize = 10;
+      const totalCount = (batchSize * 4) + 1;
+      const chunks = 5;
+
+      final tasks = await index.addDocumentsInBatches(
+        dynamicBooks(totalCount),
+        batchSize: batchSize,
+      );
+
+      expect(tasks.length, chunks);
+      await tasks.waitFor(client: client, timeout: Duration(seconds: 30));
+      final docs = await index.getDocuments();
+      expect(docs.total, totalCount);
+    });
+
     test('Add documents with primary key', () async {
       final index = client.index(randomUid());
       await index
@@ -37,6 +53,33 @@ void main() {
       expect(doc?['title'], equals('The Hobbit 2'));
     });
 
+    test('Update documents in batches', () async {
+      const batchSize = 10;
+      const chunks = 3;
+      const totalCount = (batchSize * 2) + 1;
+      final index = await createDynamicBooksIndex(count: totalCount);
+
+      final tasks = await index.updateDocumentsInBatches(
+        List.generate(
+          totalCount,
+          (index) => {
+            'book_id': index,
+            'title': 'Updated Book $index',
+          },
+        ),
+        batchSize: batchSize,
+      );
+
+      expect(tasks.length, chunks);
+      await tasks.waitFor(client: client, timeout: Duration(seconds: 30));
+      final docs = await index.getDocuments();
+      expect(docs.total, totalCount);
+      docs.results.map((element) {
+        final bookId = element['book_id'];
+        expect(element['title'], equals('Updated Book $bookId'));
+      });
+    });
+
     test('Update documents and pass a primary key', () async {
       final uid = randomUid();
       var index = client.index(uid);

test/utils/books.dart

@@ -4,6 +4,21 @@ import 'books_data.dart';
 import 'client.dart';
 import 'wait_for.dart';
 
+Future<MeiliSearchIndex> createDynamicBooksIndex({
+  String? uid,
+  required int count,
+}) async {
+  final index = client.index(uid ?? randomUid());
+  final docs = dynamicBooks(count);
+  final response = await index.addDocuments(docs).waitFor(client: client);
+
+  if (response.status != 'succeeded') {
+    throw Exception(
+        'Impossible to process test suite, the documents were not added into the index.');
+  }
+  return index;
+}
+
 Future<MeiliSearchIndex> createBooksIndex({String? uid}) async {
   return _createIndex(uid: uid);
 }
@@ -12,8 +27,10 @@ Future<MeiliSearchIndex> createNestedBooksIndex({String? uid}) async {
   return _createIndex(uid: uid, isNested: true);
 }
 
-Future<MeiliSearchIndex> _createIndex(
-    {String? uid, bool isNested = false}) async {
+Future<MeiliSearchIndex> _createIndex({
+  String? uid,
+  bool isNested = false,
+}) async {
   final index = client.index(uid ?? randomUid());
   final docs = isNested ? nestedBooks : books;
   final response = await index.addDocuments(docs).waitFor(client: client);

test/utils/books_data.dart

@@ -1,3 +1,15 @@
+List<Map<String, Object?>> dynamicBooks(int count) {
+  final tags = List.generate(4, (index) => "Tag $index");
+  return List.generate(
+    count,
+    (index) => {
+      'book_id': index,
+      'title': 'Book $index',
+      'tag': tags[index % tags.length],
+    },
+  );
+}
+
 final books = [
   {'book_id': 123, 'title': 'Pride and Prejudice', 'tag': 'Romance'},
   {'book_id': 456, 'title': 'Le Petit Prince', 'tag': 'Tale'},

test/utils/wait_for.dart

@@ -1,4 +1,5 @@
 import 'package:meilisearch/meilisearch.dart';
+import 'package:collection/collection.dart';
 
 extension TaskWaiter on Task {
   Future<Task> waitFor({
@@ -22,6 +23,41 @@ extension TaskWaiter on Task {
   }
 }
 
+extension TaskWaiterForLists on Iterable<Task> {
+  Future<List<Task>> waitFor({
+    required MeiliSearchClient client,
+    Duration timeout = const Duration(seconds: 20),
+    Duration interval = const Duration(milliseconds: 50),
+  }) async {
+    final endingTime = DateTime.now().add(timeout);
+    final originalUids = toList();
+    final remainingUids = map((e) => e.uid).whereNotNull().toList();
+    final completedTasks = <int, Task>{};
+    final statuses = ['enqueued', 'processing'];
+
+    while (DateTime.now().isBefore(endingTime)) {
+      var taskRes =
+          await client.getTasks(params: TasksQuery(uids: remainingUids));
+      final tasks = taskRes.results;
+      final completed = tasks.where((e) => !statuses.contains(e.status));
+
+      completedTasks.addEntries(completed.map((e) => MapEntry(e.uid!, e)));
+      remainingUids
+          .removeWhere((element) => completedTasks.containsKey(element));
+
+      if (remainingUids.isEmpty) {
+        return originalUids
+            .map((e) => completedTasks[e.uid])
+            .whereNotNull()
+            .toList();
+      }
+      await Future<void>.delayed(interval);
+    }
+
+    throw Exception('The tasks $originalUids timed out.');
+  }
+}
+
 extension TaskWaiterForFutures on Future<Task> {
   Future<Task> waitFor({
     required MeiliSearchClient client,
@@ -32,3 +68,14 @@ extension TaskWaiterForFutures on Future<Task> {
         .waitFor(timeout: timeout, interval: interval, client: client);
   }
 }
+
+extension TaskWaiterForFutureList on Future<Iterable<Task>> {
+  Future<List<Task>> waitFor({
+    required MeiliSearchClient client,
+    Duration timeout = const Duration(seconds: 20),
+    Duration interval = const Duration(milliseconds: 50),
+  }) async {
+    return await (await this)
+        .waitFor(timeout: timeout, interval: interval, client: client);
+  }
+}