feat(comprehend): impl connector

connectors/aws/aws-comprehend/src/main/java/io/camunda/connector/comprehend/ComprehendConnectorFunction.java

@@ -9,7 +9,11 @@
 import io.camunda.connector.api.annotation.OutboundConnector;
 import io.camunda.connector.api.outbound.OutboundConnectorContext;
 import io.camunda.connector.api.outbound.OutboundConnectorFunction;
+import io.camunda.connector.comprehend.caller.SyncComprehendCaller;
 import io.camunda.connector.comprehend.model.ComprehendRequest;
+import io.camunda.connector.comprehend.model.ComprehendRequestData;
+import io.camunda.connector.comprehend.model.ComprehendSyncRequestData;
+import io.camunda.connector.comprehend.supplier.ComprehendClientSupplier;
 import io.camunda.connector.generator.java.annotation.ElementTemplate;
 
 @OutboundConnector(
@@ -32,8 +36,30 @@
  icon = "icon.svg")
 public class ComprehendConnectorFunction implements OutboundConnectorFunction {
 
+ private ComprehendClientSupplier clientSupplier;
+
+ private SyncComprehendCaller syncComprehendCaller;
+
+ public ComprehendConnectorFunction() {
+ clientSupplier = new ComprehendClientSupplier();
+ syncComprehendCaller = new SyncComprehendCaller();
+ }
+
+ public ComprehendConnectorFunction(
+ ComprehendClientSupplier clientSupplier, SyncComprehendCaller syncComprehendCaller) {
+ this.clientSupplier = clientSupplier;
+ this.syncComprehendCaller = syncComprehendCaller;
+ }
+
  @Override
  public Object execute(OutboundConnectorContext context) throws Exception {
+ var request = context.bindVariables(ComprehendRequest.class);
+ ComprehendRequestData requestData = request.getInput();
+ if (requestData instanceof ComprehendSyncRequestData) {
+
+ return syncComprehendCaller.call(clientSupplier.getSyncClient(request),(ComprehendSyncRequestData) requestData);
+ }
  return null;
  }
+
 }

connectors/aws/aws-comprehend/src/main/java/io/camunda/connector/comprehend/caller/AsyncComprehendCaller.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH
+ * under one or more contributor license agreements. Licensed under a proprietary license.
+ * See the License.txt file for more information. You may not use this file
+ * except in compliance with the proprietary license.
+ */
+package io.camunda.connector.comprehend.caller;
+
+import com.amazonaws.services.comprehend.AmazonComprehendClient;
+import com.amazonaws.services.comprehend.model.*;
+import com.amazonaws.util.CollectionUtils;
+import io.camunda.connector.comprehend.model.ComprehendAsyncRequestData;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.commons.lang3.StringUtils;
+
+public class AsyncComprehendCaller
+ implements ComprehendCaller<StartDocumentClassificationJobResult, ComprehendAsyncRequestData> {
+
+ @Override
+ public StartDocumentClassificationJobResult call(
+ AmazonComprehendClient client, ComprehendAsyncRequestData asyncRequest) {
+ var docClassificationRequest = new StartDocumentClassificationJobRequest();
+
+ if (StringUtils.isNoneBlank(asyncRequest.clientRequestToken())) {
+ docClassificationRequest.withClientRequestToken(asyncRequest.clientRequestToken());
+ }
+
+ docClassificationRequest.withDataAccessRoleArn(asyncRequest.dataAccessRoleArn());
+
+ if (StringUtils.isNoneEmpty(asyncRequest.documentClassifierArn())) {
+ docClassificationRequest.withDocumentClassifierArn(asyncRequest.documentClassifierArn());
+ }
+
+ if (StringUtils.isNoneBlank(asyncRequest.flywheelArn())) {
+ docClassificationRequest.withFlywheelArn(asyncRequest.flywheelArn());
+ }
+
+ docClassificationRequest.withInputDataConfig(prepareInputConfig(asyncRequest));
+
+ if (StringUtils.isNoneEmpty(asyncRequest.jobName())) {
+ docClassificationRequest.withJobName(asyncRequest.jobName());
+ }
+
+ docClassificationRequest.withOutputDataConfig(prepareOutputDataConf(asyncRequest));
+
+ if (asyncRequest.tags() != null) {
+ docClassificationRequest.withTags(prepareTags(asyncRequest));
+ }
+
+ if (StringUtils.isNoneEmpty(asyncRequest.volumeKmsKeyId())) {
+ docClassificationRequest.withVolumeKmsKeyId(asyncRequest.volumeKmsKeyId());
+ }
+
+ docClassificationRequest.withVpcConfig(prepareVpcConfig(asyncRequest));
+
+ return client.startDocumentClassificationJob(docClassificationRequest);
+ }
+
+ private InputDataConfig prepareInputConfig(ComprehendAsyncRequestData request) {
+ var inputConfig =
+ new InputDataConfig()
+ .withS3Uri(request.inputS3Uri())
+ .withDocumentReaderConfig(prepareDocumentReaderConfig(request).orElse(null));
+
+ if (StringUtils.isNoneBlank(request.comprehendInputFormat().name())) {
+ inputConfig.withInputFormat(request.comprehendInputFormat().name());
+ }
+
+ return inputConfig;
+ }
+
+ private OutputDataConfig prepareOutputDataConf(ComprehendAsyncRequestData request) {
+ var outputConf = new OutputDataConfig().withS3Uri(request.outputS3Uri());
+
+ if (StringUtils.isNoneEmpty(request.outputKmsKeyId())) {
+ outputConf.withKmsKeyId(request.outputKmsKeyId());
+ }
+
+ return outputConf;
+ }
+
+ private List<Tag> prepareTags(ComprehendAsyncRequestData request) {
+ return request.tags().entrySet().stream().filter(Objects::nonNull).map(this::creatTag).toList();
+ }
+
+ private Tag creatTag(Map.Entry<String, String> entry) {
+ return new Tag().withKey(entry.getKey()).withValue(entry.getValue());
+ }
+
+ private VpcConfig prepareVpcConfig(ComprehendAsyncRequestData request) {
+ List<String> groupIds = request.securityGroupIds();
+ List<String> subnets = request.subnets();
+
+ if (CollectionUtils.isNullOrEmpty(groupIds) && CollectionUtils.isNullOrEmpty(subnets)) {
+ return null;
+ }
+
+ if (!CollectionUtils.isNullOrEmpty(groupIds) && !CollectionUtils.isNullOrEmpty(subnets)) {
+ return new VpcConfig().withSecurityGroupIds(groupIds).withSubnets(subnets);
+ } else {
+ String msg = "Or both VpcConfig fields SecurityGroupIds and Subnets or none";
+ LOGGER.warn(msg);
+ throw new IllegalArgumentException(msg);
+ }
+ }
+}

connectors/aws/aws-comprehend/src/main/java/io/camunda/connector/comprehend/caller/ComprehendCaller.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH
+ * under one or more contributor license agreements. Licensed under a proprietary license.
+ * See the License.txt file for more information. You may not use this file
+ * except in compliance with the proprietary license.
+ */
+package io.camunda.connector.comprehend.caller;
+
+import static io.camunda.connector.comprehend.model.ComprehendDocumentReadAction.TEXTRACT_ANALYZE_DOCUMENT;
+
+import com.amazonaws.AmazonWebServiceResult;
+import com.amazonaws.ResponseMetadata;
+import com.amazonaws.services.comprehend.AmazonComprehendClient;
+import com.amazonaws.services.comprehend.model.DocumentReadFeatureTypes;
+import com.amazonaws.services.comprehend.model.DocumentReaderConfig;
+import io.camunda.connector.comprehend.model.ComprehendDocumentReadMode;
+import io.camunda.connector.comprehend.model.ComprehendRequestData;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public interface ComprehendCaller<
+ T extends AmazonWebServiceResult<ResponseMetadata>, R extends ComprehendRequestData> {
+
+ Logger LOGGER = LoggerFactory.getLogger(ComprehendCaller.class);
+
+ T call(AmazonComprehendClient client, R requestData);
+
+ default Optional<DocumentReaderConfig> prepareDocumentReaderConfig(
+ ComprehendRequestData requestData) {
+ if (StringUtils.isBlank(requestData.getDocumentReadAction().name())) {
+ return Optional.empty();
+ }
+
+ var documentReaderConfig =
+ new DocumentReaderConfig()
+ .withDocumentReadAction(requestData.getDocumentReadAction().name());
+
+ if (!requestData.getDocumentReadMode().equals(ComprehendDocumentReadMode.NO_DATA)) {
+ documentReaderConfig.withDocumentReadMode(requestData.getDocumentReadMode().name());
+ }
+
+ if (requestData.getDocumentReadAction().equals(TEXTRACT_ANALYZE_DOCUMENT)) {
+ List<String> features = prepareFeatures(requestData);
+ if (features.isEmpty()) { // todo check real example
+ LOGGER.warn("DocumentReadAction: TEXTRACT_ANALYZE_DOCUMENT but features not selected.");
+ throw new IllegalArgumentException(
+ "If you chose TEXTRACT_ANALYZE_DOCUMENT as the read action,"
+ + " you must specify one feature types");
+ }
+ documentReaderConfig.withFeatureTypes(prepareFeatures(requestData));
+ }
+
+ return Optional.of(documentReaderConfig);
+
+
+ }
+
+ default List<String> prepareFeatures(ComprehendRequestData requestData) {
+ List<String> features = new ArrayList<>(2);
+ if (requestData.getFeatureTypeForms()) {
+ features.add(DocumentReadFeatureTypes.FORMS.name());
+ }
+ if (requestData.getFeatureTypeTables()) {
+ features.add(DocumentReadFeatureTypes.TABLES.name());
+ }
+ return features;
+ }
+}

connectors/aws/aws-comprehend/src/main/java/io/camunda/connector/comprehend/caller/SyncComprehendCaller.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH
+ * under one or more contributor license agreements. Licensed under a proprietary license.
+ * See the License.txt file for more information. You may not use this file
+ * except in compliance with the proprietary license.
+ */
+package io.camunda.connector.comprehend.caller;
+
+import com.amazonaws.services.comprehend.AmazonComprehendClient;
+import com.amazonaws.services.comprehend.model.ClassifyDocumentRequest;
+import com.amazonaws.services.comprehend.model.ClassifyDocumentResult;
+import io.camunda.connector.comprehend.model.ComprehendSyncRequestData;
+
+public class SyncComprehendCaller
+ implements ComprehendCaller<ClassifyDocumentResult, ComprehendSyncRequestData> {
+
+ @Override
+ public ClassifyDocumentResult call(
+ AmazonComprehendClient client, ComprehendSyncRequestData requestData) {
+ ClassifyDocumentRequest classifyDocumentRequest =
+ new ClassifyDocumentRequest()
+ .withText(requestData.text())
+ .withEndpointArn(requestData.endpointArn())
+ .withDocumentReaderConfig(prepareDocumentReaderConfig(requestData).orElse(null));
+
+ return client.classifyDocument(classifyDocumentRequest);
+ }
+}