[backend] Catalog connector ingestion data

Author: camrrx

openaev-api/src/main/java/io/openaev/runner/CatalogConnectorIngestionRunner.java

@@ -18,6 +18,7 @@
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.boot.CommandLineRunner;
  import org.springframework.stereotype.Component;
+ import org.springframework.transaction.annotation.Transactional;
 
  @Slf4j
  @RequiredArgsConstructor
@@ -52,35 +53,29 @@ public void run(String... args) {
   }
 
   private List<CatalogConnector> extractCatalog(JsonNode rootNode) {
-    List<CatalogConnector> catalogConnectors = new ArrayList<>();
-
-    JsonNode contracts = rootNode.get("contracts");
-    if (contracts != null && contracts.isArray()) {
-      for (JsonNode contract : contracts) {
-          Optional<CatalogConnector> catalog = buildCatalogConnector(contract);
-          catalog.ifPresent(catalogConnectors::add);
+      JsonNode contracts = rootNode.get("contracts");
+      if (contracts == null) {
+          // TODO throw error ?
+          return List.of();
       }
-    }
-    List<CatalogConnector> savedConnectors = catalogConnectorService.upsertAll(catalogConnectors);
 
-    if (contracts != null && contracts.isArray()) {
-      int index = 0;
-      for (JsonNode contract : contracts) {
-        CatalogConnector savedConnector = savedConnectors.get(index);
-
-        List<CatalogConnectorConfiguration> configs = buildConnectorConfigurations(contract,
- savedConnector);
-        catalogConnectorConfigurationService.upsertAll(configs);
+    List<CatalogConnector> catalogConnectorList = new ArrayList<>();
 
-        index++;
+      for (JsonNode contract : contracts) {
+          CatalogConnector catalogConnector = buildCatalogConnector(contract);
+          catalogConnectorList.add(catalogConnector);
       }
-    }
+      List<CatalogConnector> savedConnectors = catalogConnectorService.saveAll(catalogConnectorList);
 
     return savedConnectors;
   }
 
-  private Optional<CatalogConnector> buildCatalogConnector(JsonNode contract) {
-    CatalogConnector connector = new CatalogConnector();
+  private CatalogConnector buildCatalogConnector(JsonNode contract) {
+
+      CatalogConnector connector = catalogConnectorService
+              .findBySlug(contract.get("slug").asText())
+              .orElseGet(CatalogConnector::new);
+
 
     List<String> useCases = new ArrayList<>();
     JsonNode arrUseCases = contract.get("use_cases");
@@ -125,14 +120,18 @@ private Optional<CatalogConnector> buildCatalogConnector(JsonNode contract) {
         connector.setContainerType(CatalogConnector.CONNECTOR_TYPE.valueOf(containerType.trim().toUpperCase()));
     } else{
         log.error("container_type is null");
-        return Optional.empty();
+        //TODO : return empty
+        connector.setContainerType(CatalogConnector.CONNECTOR_TYPE.COLLECTOR);
     }
-    return Optional.of(connector);
+
+  Set<CatalogConnectorConfiguration> conf = buildConnectorConfigurations(contract, connector);
+  connector.setCatalogConnectorConfigurations(conf);
+
+    return connector;
   }
 
-  private List<CatalogConnectorConfiguration> buildConnectorConfigurations(JsonNode contract,
- CatalogConnector connector) {
-    List<CatalogConnectorConfiguration> configs = new ArrayList<>();
+  private Set<CatalogConnectorConfiguration> buildConnectorConfigurations(JsonNode contract, CatalogConnector connector) {
+    Set<CatalogConnectorConfiguration> configs = new HashSet<>();
 
     JsonNode schema = contract.get("config_schema");
     if (schema == null || schema.isNull()) return configs;
@@ -146,7 +145,9 @@ private List<CatalogConnectorConfiguration> buildConnectorConfigurations(JsonNod
       String key = it.next();
       JsonNode prop = properties.get(key);
 
-      CatalogConnectorConfiguration conf = new CatalogConnectorConfiguration();
+        CatalogConnectorConfiguration conf = connector.getCatalogConnectorConfigurations().stream()
+                .filter(c ->key.equals(c.getConnectorConfigurationKey())).findFirst()
+                .orElse( new CatalogConnectorConfiguration());
       conf.setCatalogConnector(connector);
       conf.setConnectorConfigurationKey(key);
 

openaev-api/src/main/java/io/openaev/service/CatalogConnectorService.java

@@ -2,7 +2,6 @@
 
 import static io.openaev.helper.StreamHelper.fromIterable;
 
-import io.openaev.database.model.AssetGroup;
 import io.openaev.database.model.CatalogConnector;
 import io.openaev.database.repository.CatalogConnectorRepository;
 import java.util.ArrayList;
@@ -11,6 +10,8 @@
 
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
 
 @RequiredArgsConstructor
 @Service
@@ -25,21 +26,21 @@ public List<CatalogConnector> saveAll(List<CatalogConnector> connectors) {
     return fromIterable(catalogConnectorRepository.saveAll(connectors));
   }
 
+  public Optional<CatalogConnector> findBySlug(String slug){
+    return catalogConnectorRepository.findBySlugWithConfigurations(slug);
+  }
+
+
+  @Transactional
   public List<CatalogConnector> upsertAll(List<CatalogConnector> connectors) {
     List<CatalogConnector> connectorsToAdd = new ArrayList<>();
 
     for (CatalogConnector connectorIncoming : connectors) {
-      Optional<CatalogConnector> connector = catalogConnectorRepository
-          .findByTitle(connectorIncoming.getTitle());
-      if (connector.isPresent()) {
-        CatalogConnector catalogConnectorFromDb = connector.get();
-        catalogConnectorFromDb.setDescription(connectorIncoming.getDescription());
-        connectorsToAdd.add(catalogConnectorRepository.save(catalogConnectorFromDb));
-      }
-      else {
-        connectorsToAdd.add(catalogConnectorRepository.save(connectorIncoming));
-      }
+      catalogConnectorRepository
+          .findByTitle(connectorIncoming.getTitle())
+          .ifPresent(existingConnector -> connectorIncoming.setId(existingConnector.getId()));
 
+      connectorsToAdd.add(catalogConnectorRepository.save(connectorIncoming));
     }
 
     return connectorsToAdd;

openaev-model/src/main/java/io/openaev/database/repository/CatalogConnectorRepository.java

@@ -3,11 +3,20 @@
 import io.openaev.database.model.CatalogConnector;
 import java.util.Optional;
 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.CrudRepository;
+import org.springframework.data.repository.query.Param;
 import org.springframework.stereotype.Repository;
 
 @Repository
 public interface CatalogConnectorRepository
     extends CrudRepository<CatalogConnector, String>, JpaSpecificationExecutor<CatalogConnector> {
   Optional<CatalogConnector> findByTitle(String title);
+
+  Optional<CatalogConnector> findBySlug(String slug);
+
+  @Query("SELECT c FROM CatalogConnector c LEFT JOIN FETCH c.catalogConnectorConfigurations WHERE c.slug = :slug")
+  Optional<CatalogConnector> findBySlugWithConfigurations(String slug);
+
+//  List<CatalogConnector> findByTitleIn(List<String> titles);
 }