Upgrade: Added the possibility to add tags to the datasets uploaded to CKAN. Both for new datasets and backups.

Signed-off-by: Ricardo Ruiz Saiz <dorm.std@gmail.com>

Author: dormstd

CKAN_API_Handler/src/main/java/net/atos/qrowd/handlers/CKAN_API_Handler.java

@@ -23,6 +23,7 @@
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
 import org.apache.http.entity.mime.MultipartEntityBuilder;
 import org.apache.http.entity.mime.content.ContentBody;
 import org.apache.http.entity.mime.content.FileBody;
@@ -37,7 +38,25 @@
 import java.io.InputStreamReader;
 import java.net.URL;
 import java.text.SimpleDateFormat;
+import java.util.ArrayList;
 import java.util.Date;
+import java.util.List;
+
+/**
+ * Copyright 2018 Atos
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 
 public class CKAN_API_Handler {
     private final Logger log = Logger.getLogger(CKAN_API_Handler.class);
@@ -147,18 +166,41 @@ public Package_ getPackageByName(String name) throws IOException {
 
     }
 
-    public void createPackage(String package_id) throws IOException{
+    public void createPackage(String package_id, String tags) throws IOException{
 
         HttpPost postRequest;
         StringBuilder sb = new StringBuilder();
         String line;
 
-        HttpEntity reqEntity = MultipartEntityBuilder.create()
-                .addPart("name",new StringBody(package_id,ContentType.TEXT_PLAIN))
-                .addPart("owner_org",new StringBody(organization_id,ContentType.TEXT_PLAIN))
-                .addPart("notes",new StringBody(package_description,ContentType.TEXT_PLAIN))
-                .addPart("private",new StringBody(package_private.toString(),ContentType.TEXT_PLAIN))
-                .build();
+        //Split <tags> by "," and for each element in the list generate a tag
+
+        String[] tagList = tags.split(",");
+        List<Tag> list = new ArrayList<>();
+        for(String tag: tagList)
+        {
+            Tag t = new Tag();
+            //Since CKAN only allows alphanumeric and _ we need to deal with illegal characters/spaces...
+            t.setName(tag.replaceAll("[\\W]+","_"));
+            list.add(t);
+        }
+
+        Package_ pack = new Package_();
+        pack.setName(package_id);
+        pack.setOwnerOrg(organization_id);
+        pack.setNotes(package_description);
+        pack.setPrivate(package_private);
+        //Set the new list of tags for the dataset
+
+        if(list.size()==0 || tags.trim().isEmpty()) {
+            pack.setTags(null);
+            pack.setNumTags(0);
+        }else {
+            pack.setTags(list);
+            pack.setNumTags(list.size());
+        }
+        Gson gson = new Gson();
+
+        StringEntity reqEntity = new StringEntity(gson.toJson(pack));
 
         postRequest = new HttpPost(HOST+"/api/3/action/package_create?use_default_schema=true");
         postRequest.setEntity(reqEntity);
@@ -189,44 +231,46 @@ public void createPackage(String package_id) throws IOException{
             log.info(sb);
         }
     }
-    public void createPackagePojo(Package_ dataset, String name) throws IOException{
+    public void createPackagePojoNoResources(Package_ dataset, String name, String tags) throws IOException{
+
+        //Split <tags> by "," and for each element in the list generate a tag
+
+        String[] tagList = tags.split(",");
+        List<Tag> list = new ArrayList<>();
+        for(String tag: tagList)
+        {
+            Tag t = new Tag();
+            //Since CKAN only allows alphanumeric and _ we need to deal with illegal characters/spaces...
+            t.setName(tag.replaceAll("[\\W]+","_"));
+            list.add(t);
+        }
 
         HttpPost postRequest;
         StringBuilder sb = new StringBuilder();
         String line;
+        //Set the new dataset name and title
+        dataset.setName(name);
+        dataset.setTitle(name);
 
-        MultipartEntityBuilder multipart = MultipartEntityBuilder.create()
-                .addPart("name",new StringBody(name,ContentType.TEXT_PLAIN));
-        // ToDo: Improve this way of handling null values in the returned dataset
-        if(dataset.getAuthor()!=null) {
-            multipart.addPart("author", new StringBody(dataset.getAuthor(), ContentType.TEXT_PLAIN));
-        }
-        if(dataset.getAuthorEmail()!=null) {
-            multipart.addPart("author_email", new StringBody(dataset.getAuthorEmail(), ContentType.TEXT_PLAIN));
-        }
-        if(dataset.getOwnerOrg()!=null) {
-            multipart.addPart("owner_org", new StringBody(dataset.getOwnerOrg(), ContentType.TEXT_PLAIN));
-        }
-        if(dataset.getNotes()!=null) {
-            multipart.addPart("notes", new StringBody(dataset.getNotes(), ContentType.TEXT_PLAIN));
-        }
-        if(dataset.getPrivate()!=null) {
-            multipart.addPart("private", new StringBody(dataset.getPrivate().toString(), ContentType.TEXT_PLAIN));
-        }
-        if(dataset.getLicenseTitle()!=null) {
-            multipart.addPart("license_title", new StringBody(dataset.getLicenseTitle(), ContentType.TEXT_PLAIN));
-        }
-        if(dataset.getLicenseId()!=null) {
-            multipart.addPart("license_id", new StringBody(dataset.getLicenseId(), ContentType.TEXT_PLAIN));
-        }
-        if(dataset.getLicenseTitle() != null){
-            multipart.addPart("license_title", new StringBody(dataset.getLicenseTitle(), ContentType.TEXT_PLAIN));
-        }
-        if(dataset.getMaintainerEmail() != null){
-            multipart.addPart("maintainer_email", new StringBody(dataset.getMaintainerEmail(), ContentType.TEXT_PLAIN));
+        //Remove identifiers of the old dataset and its resources
+        dataset.setId(null);
+        dataset.setRevisionId(null);
+        dataset.setResources(null);
+        dataset.setNumResources(null);
+
+        //Set the new list of tags for the dataset
+
+        if(list.size()==0 || tags.trim().isEmpty()) {
+            dataset.setTags(null);
+            dataset.setNumTags(0);
+        }else {
+            dataset.setTags(list);
+            dataset.setNumTags(list.size());
         }
+        Gson gson = new Gson();
 
-        HttpEntity reqEntity = multipart.build();
+        System.out.println(gson.toJson(dataset));
+        StringEntity reqEntity = new StringEntity(gson.toJson(dataset));
 
         postRequest = new HttpPost(HOST+"/api/action/package_create");
         postRequest.setEntity(reqEntity);
@@ -249,6 +293,7 @@ public void createPackagePojo(Package_ dataset, String name) throws IOException{
 
         if(statusCode!=200){
             log.error("statusCode =!=" +statusCode);
+
             log.error(sb);
         }
         else {

nifi-nifiCKANDatasetBackup-processors/src/main/java/net/atos/qrowd/processors/nifiCKANDatasetBackup/CKAN_Package_Backup.java

@@ -38,6 +38,22 @@
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 
+/**
+ * Copyright 2018 Atos
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 @EventDriven
 @SupportsBatching
 @Tags({"ckan","backup","web service","request","local"})
@@ -67,6 +83,13 @@ public class CKAN_Package_Backup extends AbstractProcessor {
             .addValidator(Validator.VALID)
             .required(true)
             .build();
+    private static final PropertyDescriptor tag_list = new PropertyDescriptor
+            .Builder().name("tag_list")
+            .displayName("Comma-separated Tag List")
+            .description("Comma-separated tag list to be set for the dataset. Only alphanumeric characters and '_' accepted")
+            .addValidator(Validator.VALID)
+            .required(false)
+            .build();
 
     private static final Relationship REL_BACKUP_CREATED = new Relationship.Builder()
             .name("BACKUP_SUCCESS")
@@ -92,6 +115,7 @@ protected void init(final ProcessorInitializationContext context) {
         descriptors.add(CKAN_url);
         descriptors.add(api_key);
         descriptors.add(package_name);
+        descriptors.add(tag_list);
 
         this.descriptors = Collections.unmodifiableList(descriptors);
 
@@ -125,6 +149,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
         String url = context.getProperty(CKAN_url).getValue();
         final String apiKey = context.getProperty(api_key).getValue();
 
+        String tagList = context.getProperty(tag_list).getValue();
+
         /* *****************
          * Main logic of the CKAN package backup:
          *  - Look in CKAN for a package with the same name as the file
@@ -154,7 +180,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
 
                 getLogger().info("Creating the package: {}", new Object[]{datasetName});
                 //Create the new timestamped package
-                ckan_api_handler.createPackagePojo(dataset,datasetName);
+                ckan_api_handler.createPackagePojoNoResources(dataset,datasetName,tagList);
 
                 //Check when the list of resources of the package is empty
                 if(resourceList.size()>0) {

nifi-nifiCKANFlowfileUploader-processors/src/main/java/net/atos/qrowd/processors/nifiCKANprocessor/CKAN_Flowfile_Uploader.java

@@ -37,6 +37,22 @@
 import java.nio.file.Paths;
 import java.util.*;
 
+/**
+ * Copyright 2018 Atos
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 @Tags({"ckan","web service","request","local"})
 @CapabilityDescription("Nifi Processor that will upload the specified flowfile to CKAN through its API, it will create the organization and package if needed.")
 @ReadsAttributes
@@ -94,6 +110,13 @@ public class CKAN_Flowfile_Uploader extends AbstractProcessor {
             .defaultValue(PRIVATE_TRUE.getValue())
             .required(true)
             .build();
+    private static final PropertyDescriptor tag_list = new PropertyDescriptor
+            .Builder().name("tag_list")
+            .displayName("Comma-separated Tag List")
+            .description("Comma-separated tag list to be set for the dataset. Only alphanumeric characters and '_' accepted")
+            .addValidator(Validator.VALID)
+            .required(false)
+            .build();
 
     private static final Relationship REL_SUCCESS = new Relationship.Builder()
             .name("SUCCESS")
@@ -118,6 +141,7 @@ protected void init(final ProcessorInitializationContext context) {
         descriptors.add(package_name);
         descriptors.add(package_description);
         descriptors.add(package_private);
+        descriptors.add(tag_list);
 
         this.descriptors = Collections.unmodifiableList(descriptors);
 
@@ -148,6 +172,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
         //This is the way to get the value of a property
         String url = context.getProperty(CKAN_url).getValue();
 
+        String tagList = context.getProperty(tag_list).getValue();
+
         String filename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
 
         String tDir = System.getProperty("java.io.tmpdir");
@@ -203,7 +229,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
                 ckan_api_handler.createOrganization();
             }
             if (!ckan_api_handler.packageExists(filenameNoExtension)) {
-                ckan_api_handler.createPackage(filenameNoExtension);
+                ckan_api_handler.createPackage(filenameNoExtension,tagList);
             }
             if(ckan_api_handler.createOrUpdateResource(file.toFile().toString())) {
                 getLogger().info("File tried to be uploaded to CKAN: {}", new Object[]{file.toFile().toString()});