This topic was automatically generated from Slack. You can find the original thread here.
Hi I am doing Bulk operations with Shopify, requiring to execute every day a mutation - bulkOperationRunQuery graphql query to Shopify, and then shopify send me a webhook informing my query is ready to download (JSONL) which trigger a new workflow. In the webhook, I receive the reference of the graphql that Shopify generated in my Bulk operation.
The problem is that I generate Bulk Operations from other workflow, and from other system also.
How do you suggest to handle that ? Should I register the graphql reference generated during the Bulk Operations to a database, and then, each webhook should check if this webhook corresponds to its process ?
Based on your description, it seems you are looking for a way to track and associate the bulk operation GraphQL queries you initiate with the corresponding webhooks that Shopify sends when the operation is ready. Here’s a suggested approach using Pipedream’s built-in data stores: 1. Register the GraphQL Reference: When you initiate a bulk operation from your workflow or another system, store the GraphQL reference in a Pipedream data store. This will act as a registry for all initiated bulk operations. 2. Webhook Trigger: Set up a Pipedream workflow with a webhook trigger to receive notifications from Shopify. 3. Check the Registry: In the webhook-triggered workflow, add a step to check the Pipedream data store for the GraphQL reference received in the webhook payload. This will help you verify if the webhook corresponds to a process you initiated. 4. Process the Data: If the reference is found in the data store, proceed with your workflow logic to download and process the JSONL file as required. 5. Clean Up: After processing the data, consider removing the reference from the data store to keep it clean and prevent it from growing indefinitely. Here’s a simplified example of how you might write a Pipedream component to check the data store: export default defineComponent({ props: { data: { type: "data_store" }, }, async run({ steps, $ }) { const graphqlRef = steps.trigger.event.body.graphql_reference; const exists = await this.data.has(graphqlRef); if (exists) { // Proceed with processing since the reference exists in the data store // Your logic here } else { // The reference does not exist, you may want to log this or take other actions console.log("Received webhook for unknown GraphQL reference"); } }, }); Remember to replace steps.trigger.event.body.graphql_reference with the actual path to the GraphQL reference in the webhook payload. This approach should help you manage and track your bulk operations without the need for an external database. If you have any further questions or need more detailed assistance, please let me know!
Hi ! Instead of configuring several Pipedream workflow receiving all the webhooks, I prefer to have one workflow who receive the webhook, check the data store and then trigger the related workflow. How should I do that ? Can you provide example for Node.JS ? Thank you !
To achieve your goal of having a single Pipedream workflow that receives all webhooks, checks a data store, and then triggers related workflows, you can follow these steps: 1. Set up a single workflow with an HTTP/Webhook trigger to receive incoming webhooks. 2. Use a data store to keep track of which workflow should be triggered based on the webhook data. 3. Use the Pipedream API or a custom logic to trigger the related workflow. Here’s an example in Node.js: import { axios } from "@pipedream/platform"; export default defineComponent({ props: { data: { type: "data_store" }, }, async run({ steps, $ }) { // Assume the incoming webhook data has an 'eventType' that determines the workflow const eventType = steps.trigger.event.body.eventType; // Check the data store for the workflow ID associated with this event type const workflowId = await this.data.get(eventType); // If a workflow ID is found, trigger the related workflow if (workflowId) { const response = await axios($, { method: "POST", url: `https://api.pipedream.com/v1/workflows/${workflowId}/dispatches`, headers: { "Authorization": `Bearer YOUR_API_KEY`, "Content-Type": "application/json", }, data: { // Include any data you want to pass to the triggered workflow data: steps.trigger.event.body, }, }); // Log the response from the API call console.log(response); } else { // Handle the case where no workflow is associated with the event type console.log(`No workflow found for event type: ${eventType}`); } }, }); Please replace YOUR_API_KEY with your Pipedream API key. This example assumes that you have stored the mapping of event types to workflow IDs in your data store, where the key is the eventType and the value is the workflow ID. Remember to handle the case where the eventType does not match any workflow ID in your data store, and ensure that your workflows are set up to be triggered via API dispatches. Keep in mind that triggering workflows programmatically via the Pipedream API is subject to the permissions and rate limits associated with your account.
To adapt the code for a Shopify Bulk Operation, you would typically use the Shopify Admin API to perform bulk operations. However, without specific details on the operation you want to perform or the data you want to handle, I can only provide a general example. Here’s a Node.js Pipedream component that initiates a Shopify Bulk Operation to query products: ```
import { axios } from “@pipedream/platform”; export default defineComponent({ props: { shopify: { type: “app”, app: “shopify”, }, }, async run({ steps, $ }) { const query = mutation { bulkOperationRunQuery( query: """ { products { edges { node { id title handle } } } } """ ) { bulkOperation { id status } userErrors { field message } } } ; const response = await axios($, { method: “POST”, url: https://${this.shopify.$auth.shop_id}.[myshopify.com/admin/api/2021-10/graphql.json](http://myshopify.com/admin/api/2021-10/graphql.json), headers: { “X-Shopify-Access-Token”: this.shopify.$auth.oauth_access_token, “Content-Type”: “application/json”, }, data: { query: query, }, }); return response; }, });