Added API to retrieve expanded golden records for UID list and optimized backup logic

Author: sushantpatil1214

JeMPI_Apps/JeMPI_BackupRestoreAPI/src/main/java/org/jembi/jempi/backuprestoreapi/Ask.java

@@ -9,6 +9,7 @@
 import org.jembi.jempi.shared.models.ApiModels;
 import org.jembi.jempi.shared.models.RestoreGoldenRecords;
 
+import java.util.List;
 import java.util.concurrent.CompletionStage;
 
 public final class Ask {
@@ -42,6 +43,18 @@ static CompletionStage<BackEnd.GetExpandedGoldenRecordResponse> getExpandedGolde
       return stage.thenApply(response -> response);
    }
 
+   static CompletionStage<BackEnd.GetExpandedGoldenRecordsResponse> getExpandedGoldenRecords(
+           final ActorSystem<Void> actorSystem,
+           final ActorRef<BackEnd.Event> backEnd,
+           final List<String> gidList) {
+      CompletionStage<BackEnd.GetExpandedGoldenRecordsResponse> stage = AskPattern
+              .ask(backEnd,
+                      replyTo -> new BackEnd.GetExpandedGoldenRecordsRequest(replyTo, gidList),
+                      java.time.Duration.ofSeconds(GlobalConstants.TIMEOUT_DGRAPH_QUERY_SECS),
+                      actorSystem.scheduler());
+      return stage.thenApply(response -> response);
+   }
+
    static CompletionStage<BackEnd.PostGoldenRecordResponse> postGoldenRecord(
            final ActorSystem<Void> actorSystem,
            final ActorRef<BackEnd.Event> backEnd,

JeMPI_Apps/JeMPI_BackupRestoreAPI/src/main/java/org/jembi/jempi/backuprestoreapi/BackEnd.java

@@ -12,6 +12,7 @@
 import org.jembi.jempi.libmpi.MpiServiceError;
 import org.jembi.jempi.shared.models.*;
 
+import java.util.Collections;
 import java.util.List;
 
 public final class BackEnd extends AbstractBehavior<BackEnd.Event> {
@@ -103,6 +104,7 @@ public Receive<Event> actor() {
       ReceiveBuilder<Event> builder = newReceiveBuilder();
       return builder.onMessage(GetGidsAllRequest.class, this::getGidsAllHandler)
                     .onMessage(GetExpandedGoldenRecordRequest.class, this::getExpandedGoldenRecordHandler)
+                    .onMessage(GetExpandedGoldenRecordsRequest.class, this::getExpandedGoldenRecordsHandler)
                     .onMessage(PostGoldenRecordRequest.class, this::postGoldenRecordRequestHandler)
                     .build();
    }
@@ -135,6 +137,26 @@ private Behavior<Event> getExpandedGoldenRecordHandler(final GetExpandedGoldenRe
       return Behaviors.same();
    }
 
+   private Behavior<Event> getExpandedGoldenRecordsHandler(final GetExpandedGoldenRecordsRequest request) {
+      List<ExpandedGoldenRecord> goldenRecords = null;
+      try {
+         goldenRecords = libMPI.findExpandedGoldenRecords(request.goldenIds);
+      } catch (Exception exception) {
+         LOGGER.error("libMPI.findExpandedGoldenRecords failed for goldenIds: {} with error: {}",
+                 request.goldenIds,
+                 exception.getMessage());
+      }
+
+      if (goldenRecords == null) {
+         request.replyTo.tell(new GetExpandedGoldenRecordsResponse(Either.left(new MpiServiceError.GoldenIdDoesNotExistError(
+                 "Golden Records do not exist",
+                 Collections.singletonList(request.goldenIds).toString()))));
+      } else {
+         request.replyTo.tell(new GetExpandedGoldenRecordsResponse(Either.right(goldenRecords)));
+      }
+      return Behaviors.same();
+   }
+
    private Behavior<Event> postGoldenRecordRequestHandler(final PostGoldenRecordRequest request) {
       String goldenRecords = null;
       try {
@@ -169,6 +191,13 @@ public record GetExpandedGoldenRecordResponse(Either<MpiGeneralError, ExpandedGo
          implements EventResponse {
    }
 
+   public record GetExpandedGoldenRecordsRequest(
+           ActorRef<GetExpandedGoldenRecordsResponse> replyTo,
+           List<String> goldenIds) implements Event { }
+
+   public record GetExpandedGoldenRecordsResponse(
+           Either<MpiGeneralError, List<ExpandedGoldenRecord>> expandedGoldenRecords) implements EventResponse { }
+
    public record PostGoldenRecordRequest(
            ActorRef<PostGoldenRecordResponse> replyTo,
            RestoreGoldenRecords goldenRecord) implements Event {

JeMPI_Apps/JeMPI_BackupRestoreAPI/src/main/java/org/jembi/jempi/backuprestoreapi/Routes.java

@@ -91,13 +91,38 @@ private static Route getExpandedGoldenRecord(
                                            }));
     }
 
+    private static Route getExpandedGoldenRecordsForUidList(
+            final ActorSystem<Void> actorSystem,
+            final ActorRef<BackEnd.Event> backEnd) {
+        return entity(Jackson.unmarshaller(ApiModels.ApiExpandedGoldenRecordsParameterList.class),
+                request -> onComplete(Ask.getExpandedGoldenRecords(actorSystem, backEnd, request.uidList()),
+                        result -> {
+                            if (!result.isSuccess()) {
+                                final var e = result.failed().get();
+                                LOGGER.error(e.getLocalizedMessage(), e);
+                                return mapError(new MpiServiceError.InternalError(e.getLocalizedMessage()));
+                            }
+                            return result.get()
+                                    .expandedGoldenRecords()
+                                    .mapLeft(MapError::mapError)
+                                    .fold(error -> error,
+                                            expandedGoldenRecords -> complete(StatusCodes.OK,
+                                                    expandedGoldenRecords.stream()
+                                                            .map(ApiModels.ApiExpandedGoldenRecord::fromExpandedGoldenRecord)
+                                                            .toList(),
+                                                    JSON_MARSHALLER));
+                        }));
+    }
+
     public static Route createCoreAPIRoutes(
             final ActorSystem<Void> actorSystem,
             final ActorRef<BackEnd.Event> backEnd
     ) {
         return concat(post(() -> concat(
                         path(GlobalConstants.SEGMENT_POST_EXPANDED_GOLDEN_RECORD,
                                 () -> Routes.getExpandedGoldenRecord(actorSystem, backEnd)),
+                        path(GlobalConstants.SEGMENT_POST_EXPANDED_GOLDEN_RECORDS_FOR_UID_LIST,
+                                () -> Routes.getExpandedGoldenRecordsForUidList(actorSystem, backEnd)),
                         path(GlobalConstants.SEGMENT_POST_GOLDEN_RECORD_RESTORE,
                                 () -> Routes.postGoldenRecord(actorSystem, backEnd))
                 )),

devops/linux/docker/backup_restore/dgraph-backup.py

@@ -22,15 +22,15 @@ def create_folder_if_not_exists(folder_path):
 
 # Function to fetch data for a single ID
 def fetch_data_for_id(gid):
-    get_expanded_golden_record = f'http://{host}:{port}/JeMPI/expandedGoldenRecord'
-    payload = json.dumps({"gid": gid})
+    get_expanded_golden_record = f'http://{host}:{port}/JeMPI/expandedGoldenRecords'
+    payload = json.dumps({"uidList": gid})
     headers = {'Content-Type': 'application/json'}
     response = requests.post(get_expanded_golden_record, headers=headers, data=payload)
-    if response.status_code == 200:
-        return response.json()
-    else:
-        print(f'Error fetching data for ID {gid}: {response.status_code}')
-    return None
+    return response.json() if response.status_code == 200 else None
+
+def chunks(lst, n):
+    for i in range(0, len(lst), n):
+        yield lst[i:i + n]
 
 # Function to backup Dgraph data
 def backup_dgraph_data():
@@ -40,20 +40,22 @@ def backup_dgraph_data():
     if response.status_code == 200:
         new_golden_records = response.json()
         backup_data = []
-        for gid in new_golden_records.get("records", []):
-            try:
-                golden_record_data = fetch_data_for_id(gid)
-                if golden_record_data:
+        gids = new_golden_records.get("records")
+        chunk_size = 100
+        for gid_chunk in chunks(gids, chunk_size):
+            golden_records_data = fetch_data_for_id(gid_chunk)
+            if golden_records_data:
+                for golden_record_data in golden_records_data:
                     backup_data.append(golden_record_data)
-            except Exception as e:
-                print(f'Error processing ID {gid}: {e}')
+
         file_name = f'dgraph_backup_{current_datetime}.json'
         print(f'Total {str(len(backup_data))} Golden records backed up.')
         backup_path_folder = create_backup_json(backup_data, file_name)
         print(f'All data saved to {backup_path_folder + "/" + file_name}')
     else:
         print('Failed to retrieve list of IDs from the API')
 
+
 def create_backup_json(backup_data, file_name):
     backup_path_folder = os.path.join(backup_path, current_datetime)
     create_folder_if_not_exists(backup_path_folder)