turcsanyip commented on code in PR #6303: URL: https://github.com/apache/nifi/pull/6303#discussion_r962965217
########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/pom.xml: ########## @@ -0,0 +1,100 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You 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. +--> +<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xmlns="http://maven.apache.org/POM/4.0.0"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>nifi-shopify-bundle</artifactId> + <groupId>org.apache.nifi</groupId> + <version>1.18.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>nifi-shopify-processors</artifactId> + + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-api</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-utils</artifactId> + <version>1.18.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>commons-io</groupId> + <artifactId>commons-io</artifactId> + </dependency> + <dependency> + <groupId>com.squareup.okhttp3</groupId> + <artifactId>mockwebserver</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-web-client-provider-api</artifactId> + <version>1.18.0-SNAPSHOT</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service-api</artifactId> + </dependency> Review Comment: This dependency is not used and can be removed. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/pom.xml: ########## @@ -0,0 +1,100 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You 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. +--> +<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xmlns="http://maven.apache.org/POM/4.0.0"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>nifi-shopify-bundle</artifactId> + <groupId>org.apache.nifi</groupId> + <version>1.18.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>nifi-shopify-processors</artifactId> + + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-api</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-utils</artifactId> + <version>1.18.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>commons-io</groupId> + <artifactId>commons-io</artifactId> + </dependency> + <dependency> + <groupId>com.squareup.okhttp3</groupId> + <artifactId>mockwebserver</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-web-client-provider-api</artifactId> + <version>1.18.0-SNAPSHOT</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service-api</artifactId> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </dependency> + <!-- Test dependencies --> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-mock</artifactId> + <version>1.18.0-SNAPSHOT</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-web-client-provider-service</artifactId> + <version>1.18.0-SNAPSHOT</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service</artifactId> + <scope>test</scope> + </dependency> Review Comment: This dependency is not used and can be removed. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") + .required(false) + .identifiesControllerService(WebClientServiceProvider.class) + .build(); + + static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder() + .name("api-url") + .displayName("API URL") + .description("The API URL of the Custom Shopify App") Review Comment: This property expects only the "hostname" part of the URL, not the whole API URL. E.g. `nifistore.myshopify.com` instead of `https://nifistore.myshopify.com/admin/api`. So I think `Store Domain` / `The domain of the Shopify store` would be a better name / description for it. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") + .required(false) + .identifiesControllerService(WebClientServiceProvider.class) + .build(); + + static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder() + .name("api-url") + .displayName("API URL") + .description("The API URL of the Custom Shopify App") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor ACCESS_TOKEN = new PropertyDescriptor.Builder() + .name("access-token") + .displayName("Admin API Access Token") + .description("The Admin API Access Token of the Custom Shopify App") + .required(true) + .sensitive(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor API_VERSION = new PropertyDescriptor.Builder() + .name("api-version") + .displayName("API Version") + .description("The used REST API version") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .defaultValue("2022-10") + .build(); + + static final PropertyDescriptor RESOURCE_TYPE = new PropertyDescriptor.Builder() + .name("resource-type") + .displayName("Resource Type") + .description("Shopify resource type") + .required(true) + .allowableValues(ResourceDirectory.getCategories()) + .build(); + + static final Relationship REL_SUCCESS = new Relationship.Builder() + .name("success") + .description("For FlowFiles created as a result of a successful query.") + .build(); + + private static final Map<ResourceType, PropertyDescriptor> propertyMap = new EnumMap<>(ResourceType.class); + private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = createPropertyDescriptors(); + + private static List<PropertyDescriptor> createPropertyDescriptors() { + final List<PropertyDescriptor> resourceDescriptors = Arrays.stream(ResourceType.values()) + .map(resourceType -> { + final PropertyDescriptor resourceDescriptor = new PropertyDescriptor.Builder() + .name(resourceType.getValue()) + .displayName(resourceType.getDisplayName()) + .description(resourceType.getDescription()) + .required(true) + .dependsOn(RESOURCE_TYPE, resourceType.getValue()) + .allowableValues(ResourceDirectory.getResourcesAsAllowableValues(resourceType)) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .build(); + propertyMap.put(resourceType, resourceDescriptor); + return resourceDescriptor; + }) + .collect(Collectors.toList()); + final List<PropertyDescriptor> propertyDescriptors = new ArrayList<>(Arrays.asList( + WEB_CLIENT_PROVIDER, Review Comment: Similar to `GetHubSpot`, this property can go to the bottom of the property list. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-nar/pom.xml: ########## @@ -0,0 +1,46 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>nifi-shopify-bundle</artifactId> + <groupId>org.apache.nifi</groupId> + <version>1.18.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>nifi-shopify-nar</artifactId> + + <packaging>nar</packaging> + <properties> + <maven.javadoc.skip>true</maven.javadoc.skip> + <source.skip>true</source.skip> + </properties> + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-shopify-processors</artifactId> + <version>1.18.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-standard-services-api-nar</artifactId> + <version>1.18.0-SNAPSHOT</version> + <type>nar</type> + </dependency> + </dependencies> +</project> Review Comment: Please add LICENSE / NOTICE files for the NAR bundle. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") + .required(false) + .identifiesControllerService(WebClientServiceProvider.class) + .build(); + + static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder() + .name("api-url") + .displayName("API URL") + .description("The API URL of the Custom Shopify App") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor ACCESS_TOKEN = new PropertyDescriptor.Builder() + .name("access-token") + .displayName("Admin API Access Token") + .description("The Admin API Access Token of the Custom Shopify App") + .required(true) + .sensitive(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor API_VERSION = new PropertyDescriptor.Builder() + .name("api-version") + .displayName("API Version") + .description("The used REST API version") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .defaultValue("2022-10") + .build(); + + static final PropertyDescriptor RESOURCE_TYPE = new PropertyDescriptor.Builder() + .name("resource-type") + .displayName("Resource Type") + .description("Shopify resource type") + .required(true) + .allowableValues(ResourceDirectory.getCategories()) + .build(); + + static final Relationship REL_SUCCESS = new Relationship.Builder() + .name("success") + .description("For FlowFiles created as a result of a successful query.") + .build(); + + private static final Map<ResourceType, PropertyDescriptor> propertyMap = new EnumMap<>(ResourceType.class); + private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = createPropertyDescriptors(); + + private static List<PropertyDescriptor> createPropertyDescriptors() { + final List<PropertyDescriptor> resourceDescriptors = Arrays.stream(ResourceType.values()) + .map(resourceType -> { + final PropertyDescriptor resourceDescriptor = new PropertyDescriptor.Builder() + .name(resourceType.getValue()) + .displayName(resourceType.getDisplayName()) + .description(resourceType.getDescription()) + .required(true) + .dependsOn(RESOURCE_TYPE, resourceType.getValue()) + .allowableValues(ResourceDirectory.getResourcesAsAllowableValues(resourceType)) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .build(); + propertyMap.put(resourceType, resourceDescriptor); + return resourceDescriptor; + }) + .collect(Collectors.toList()); + final List<PropertyDescriptor> propertyDescriptors = new ArrayList<>(Arrays.asList( + WEB_CLIENT_PROVIDER, + API_URL, + ACCESS_TOKEN, + API_VERSION, + RESOURCE_TYPE + )); + propertyDescriptors.addAll(resourceDescriptors); + return Collections.unmodifiableList(propertyDescriptors); + } + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final JsonFactory JSON_FACTORY = OBJECT_MAPPER.getFactory(); + private static final int TOO_MANY_REQUESTS = 429; + + private volatile ShopifyRestService shopifyRestService; + private volatile ShopifyResource shopifyResource; + private volatile String resourceName; + + @OnScheduled + public void onScheduled(final ProcessContext context) { + final WebClientServiceProvider webClientServiceProvider = + context.getProperty(WEB_CLIENT_PROVIDER).asControllerService(WebClientServiceProvider.class); + final WebClientService webClientService = webClientServiceProvider.getWebClientService(); + final HttpUriBuilder uriBuilder = webClientServiceProvider.getHttpUriBuilder(); + + final String apiVersion = context.getProperty(API_VERSION).getValue(); + final String baseUrl = context.getProperty(API_URL).getValue(); + final String accessToken = context.getProperty(ACCESS_TOKEN).getValue(); + + final String category = context.getProperty(RESOURCE_TYPE).getValue(); + final ResourceType resourceType = ResourceType.valueOf(category); + resourceName = context.getProperty(propertyMap.get(resourceType)).getValue(); + + shopifyResource = ResourceDirectory.getResourceTypeDto(resourceType, resourceName); + + shopifyRestService = + getShopifyRestService(webClientService, uriBuilder, apiVersion, baseUrl, accessToken, resourceName, + shopifyResource.getIncrementalLoadingParameter()); + } + + ShopifyRestService getShopifyRestService(final WebClientService webClientService, final HttpUriBuilder uriBuilder, + final String apiVersion, final String baseUrl, final String accessToken, final String resourceName, + final IncrementalLoadingParameter incrementalLoadingParameter) { + return new ShopifyRestService( + webClientService, + uriBuilder, + apiVersion, + baseUrl, + accessToken, + resourceName, + incrementalLoadingParameter + ); + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return PROPERTY_DESCRIPTORS; + } + + @Override + public Set<Relationship> getRelationships() { + final Set<Relationship> relationships = new HashSet<>(); + relationships.add(REL_SUCCESS); + return relationships; Review Comment: Please initialize the set once (static field and initializer) instead of creating it every time. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/pom.xml: ########## @@ -0,0 +1,100 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You 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. +--> +<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xmlns="http://maven.apache.org/POM/4.0.0"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>nifi-shopify-bundle</artifactId> + <groupId>org.apache.nifi</groupId> + <version>1.18.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>nifi-shopify-processors</artifactId> + + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-api</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-utils</artifactId> + <version>1.18.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>commons-io</groupId> + <artifactId>commons-io</artifactId> + </dependency> + <dependency> + <groupId>com.squareup.okhttp3</groupId> + <artifactId>mockwebserver</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-web-client-provider-api</artifactId> + <version>1.18.0-SNAPSHOT</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service-api</artifactId> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </dependency> + <!-- Test dependencies --> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-mock</artifactId> + <version>1.18.0-SNAPSHOT</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-web-client-provider-service</artifactId> + <version>1.18.0-SNAPSHOT</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-proxy-configuration-api</artifactId> + <scope>test</scope> + </dependency> Review Comment: This dependency is not used and can be removed. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") + .required(false) + .identifiesControllerService(WebClientServiceProvider.class) + .build(); + + static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder() + .name("api-url") + .displayName("API URL") + .description("The API URL of the Custom Shopify App") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor ACCESS_TOKEN = new PropertyDescriptor.Builder() + .name("access-token") + .displayName("Admin API Access Token") + .description("The Admin API Access Token of the Custom Shopify App") Review Comment: The same Display Name and Description should be used as in case of `GetHubSpot`. ```suggestion .displayName("Access Token") .description("Access Token to authenticate requests") ``` ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") + .required(false) + .identifiesControllerService(WebClientServiceProvider.class) + .build(); + + static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder() + .name("api-url") + .displayName("API URL") + .description("The API URL of the Custom Shopify App") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor ACCESS_TOKEN = new PropertyDescriptor.Builder() + .name("access-token") + .displayName("Admin API Access Token") + .description("The Admin API Access Token of the Custom Shopify App") + .required(true) + .sensitive(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor API_VERSION = new PropertyDescriptor.Builder() + .name("api-version") + .displayName("API Version") + .description("The used REST API version") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .defaultValue("2022-10") + .build(); + + static final PropertyDescriptor RESOURCE_TYPE = new PropertyDescriptor.Builder() + .name("resource-type") + .displayName("Resource Type") Review Comment: `Resource` sounds to me a bit technical and related to the REST API. I would consider to use `Object Category` for this property and `Object Type` for the dependant properties. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") + .required(false) + .identifiesControllerService(WebClientServiceProvider.class) + .build(); + + static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder() + .name("api-url") + .displayName("API URL") + .description("The API URL of the Custom Shopify App") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor ACCESS_TOKEN = new PropertyDescriptor.Builder() + .name("access-token") + .displayName("Admin API Access Token") + .description("The Admin API Access Token of the Custom Shopify App") + .required(true) + .sensitive(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor API_VERSION = new PropertyDescriptor.Builder() + .name("api-version") + .displayName("API Version") + .description("The used REST API version") Review Comment: ```suggestion .description("The Shopify REST API version") ``` ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") + .required(false) Review Comment: In my understanding it should be required. ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") Review Comment: Please remove "NiFi" prefix. The description could be the same as in case of `GetHubSpot` processor. ```suggestion .displayName("Web Client Service Provider") .description("Controller service for HTTP client operations") ``` ########## nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java: ########## @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.shopify; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter; +import org.apache.nifi.processors.shopify.model.ResourceDirectory; +import org.apache.nifi.processors.shopify.model.ResourceType; +import org.apache.nifi.processors.shopify.model.ShopifyResource; +import org.apache.nifi.processors.shopify.rest.ShopifyRestService; +import org.apache.nifi.web.client.api.HttpResponseEntity; +import org.apache.nifi.web.client.api.HttpResponseStatus; +import org.apache.nifi.web.client.api.HttpUriBuilder; +import org.apache.nifi.web.client.api.WebClientService; +import org.apache.nifi.web.client.provider.api.WebClientServiceProvider; + +@PrimaryNodeOnly +@TriggerSerially +@TriggerWhenEmpty +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Tags({"shopify"}) +@Stateful(scopes = Scope.CLUSTER, description = + "For a few resources the processors support incremental loading. The list of the resources with the supported parameters" + + + "can be found in additional details. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is" + + + " selected, the new node can pick up where the previous node left off, without duplicating the data.") +@CapabilityDescription("Retrieves object from a custom Shopify store. The processor yield time must be set to the account's rate limit accordingly.") +public class GetShopify extends AbstractProcessor { + + public static final PropertyDescriptor WEB_CLIENT_PROVIDER = new PropertyDescriptor.Builder() + .name("web-client-service-provider") + .displayName("NiFi Web Client Service Provider") + .description("NiFi Web Client Service Provider to make HTTP calls and build URIs") + .required(false) + .identifiesControllerService(WebClientServiceProvider.class) + .build(); + + static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder() + .name("api-url") + .displayName("API URL") + .description("The API URL of the Custom Shopify App") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor ACCESS_TOKEN = new PropertyDescriptor.Builder() + .name("access-token") + .displayName("Admin API Access Token") + .description("The Admin API Access Token of the Custom Shopify App") + .required(true) + .sensitive(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor API_VERSION = new PropertyDescriptor.Builder() + .name("api-version") + .displayName("API Version") + .description("The used REST API version") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .defaultValue("2022-10") + .build(); + + static final PropertyDescriptor RESOURCE_TYPE = new PropertyDescriptor.Builder() + .name("resource-type") + .displayName("Resource Type") + .description("Shopify resource type") + .required(true) + .allowableValues(ResourceDirectory.getCategories()) + .build(); + + static final Relationship REL_SUCCESS = new Relationship.Builder() + .name("success") + .description("For FlowFiles created as a result of a successful query.") + .build(); + + private static final Map<ResourceType, PropertyDescriptor> propertyMap = new EnumMap<>(ResourceType.class); + private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = createPropertyDescriptors(); + + private static List<PropertyDescriptor> createPropertyDescriptors() { + final List<PropertyDescriptor> resourceDescriptors = Arrays.stream(ResourceType.values()) + .map(resourceType -> { + final PropertyDescriptor resourceDescriptor = new PropertyDescriptor.Builder() + .name(resourceType.getValue()) + .displayName(resourceType.getDisplayName()) + .description(resourceType.getDescription()) + .required(true) + .dependsOn(RESOURCE_TYPE, resourceType.getValue()) + .allowableValues(ResourceDirectory.getResourcesAsAllowableValues(resourceType)) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .build(); + propertyMap.put(resourceType, resourceDescriptor); + return resourceDescriptor; + }) + .collect(Collectors.toList()); + final List<PropertyDescriptor> propertyDescriptors = new ArrayList<>(Arrays.asList( + WEB_CLIENT_PROVIDER, + API_URL, + ACCESS_TOKEN, + API_VERSION, + RESOURCE_TYPE + )); + propertyDescriptors.addAll(resourceDescriptors); + return Collections.unmodifiableList(propertyDescriptors); + } + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final JsonFactory JSON_FACTORY = OBJECT_MAPPER.getFactory(); + private static final int TOO_MANY_REQUESTS = 429; + + private volatile ShopifyRestService shopifyRestService; + private volatile ShopifyResource shopifyResource; + private volatile String resourceName; + + @OnScheduled + public void onScheduled(final ProcessContext context) { + final WebClientServiceProvider webClientServiceProvider = + context.getProperty(WEB_CLIENT_PROVIDER).asControllerService(WebClientServiceProvider.class); + final WebClientService webClientService = webClientServiceProvider.getWebClientService(); + final HttpUriBuilder uriBuilder = webClientServiceProvider.getHttpUriBuilder(); + + final String apiVersion = context.getProperty(API_VERSION).getValue(); + final String baseUrl = context.getProperty(API_URL).getValue(); + final String accessToken = context.getProperty(ACCESS_TOKEN).getValue(); + + final String category = context.getProperty(RESOURCE_TYPE).getValue(); + final ResourceType resourceType = ResourceType.valueOf(category); + resourceName = context.getProperty(propertyMap.get(resourceType)).getValue(); + + shopifyResource = ResourceDirectory.getResourceTypeDto(resourceType, resourceName); + + shopifyRestService = + getShopifyRestService(webClientService, uriBuilder, apiVersion, baseUrl, accessToken, resourceName, + shopifyResource.getIncrementalLoadingParameter()); + } + + ShopifyRestService getShopifyRestService(final WebClientService webClientService, final HttpUriBuilder uriBuilder, + final String apiVersion, final String baseUrl, final String accessToken, final String resourceName, + final IncrementalLoadingParameter incrementalLoadingParameter) { + return new ShopifyRestService( + webClientService, + uriBuilder, + apiVersion, + baseUrl, + accessToken, + resourceName, + incrementalLoadingParameter + ); + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return PROPERTY_DESCRIPTORS; + } + + @Override + public Set<Relationship> getRelationships() { + final Set<Relationship> relationships = new HashSet<>(); + relationships.add(REL_SUCCESS); + return relationships; + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + final StateMap state = getState(context); + final String fromDateTime = state.get(resourceName); + + final HttpResponseEntity response = shopifyRestService.getShopifyObjects(fromDateTime); + final AtomicInteger objectCountHolder = new AtomicInteger(); + + if (response.statusCode() == HttpResponseStatus.OK.getCode()) { + FlowFile flowFile = session.create(); + flowFile = session.write(flowFile, parseHttpResponse(response, objectCountHolder)); + if (objectCountHolder.get() > 0) { + session.transfer(flowFile, REL_SUCCESS); + } else { + getLogger().debug("Empty response when requested Shopify resource: [{}]", resourceName); + session.remove(flowFile); + } + } else if (response.statusCode() >= 400) { + if (response.statusCode() == TOO_MANY_REQUESTS) { + context.yield(); + throw new ProcessException(String.format( + "Rate limit exceeded, yielding before retrying request. HTTP %d error for requested URI [%s]", + response.statusCode(), resourceName)); + } else { + context.yield(); + getLogger().warn("HTTP {} error for requested Shopify resource [{}]", response.statusCode(), + resourceName); + } + } + + Map<String, String> newState = new HashMap<>(state.toMap()); + if (shopifyResource.getIncrementalLoadingParameter() != IncrementalLoadingParameter.NONE) { + newState.put(shopifyRestService.getResourceName(), getCurrentExecutionTime()); Review Comment: I'm afraid it is not enough to get the current time at the end of the processing and store it because new items may be created after the REST call but before this point. I think we need to get the current time at the beginning and use it as the upper limit (`created_at_max`) when retrieving objects. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
