Skip to content

Commit

Permalink
feat(Execute Workflow Node): Run once for each item mode (#7289)
Browse files Browse the repository at this point in the history
Github issue / Community forum post (link here to close automatically):
  • Loading branch information
michael-radency committed Oct 6, 2023
1 parent 597669a commit c8c14ca
Show file tree
Hide file tree
Showing 3 changed files with 149 additions and 54 deletions.
133 changes: 79 additions & 54 deletions packages/nodes-base/nodes/ExecuteWorkflow/ExecuteWorkflow.node.ts
@@ -1,14 +1,13 @@
import { readFile as fsReadFile } from 'fs/promises';

import { NodeOperationError } from 'n8n-workflow';
import type {
IExecuteFunctions,
IExecuteWorkflowInfo,
INodeExecutionData,
INodeType,
INodeTypeDescription,
IWorkflowBase,
} from 'n8n-workflow';
import { NodeOperationError } from 'n8n-workflow';

import { getWorkflowInfo } from './GenericFunctions';
import { generatePairedItemData } from '../../utils/utilities';

export class ExecuteWorkflow implements INodeType {
description: INodeTypeDescription = {
Expand Down Expand Up @@ -83,7 +82,9 @@ export class ExecuteWorkflow implements INodeType {
},
default: '',
required: true,
description: 'The workflow to execute',
hint: 'Can be found in the URL of the workflow',
description:
"Note on using an expression here: if this node is set to run once with all items, they will all be sent to the <em>same</em> workflow. That workflow's ID will be calculated by evaluating the expression for the <strong>first input item</strong>.",
},

// ----------------------------------
Expand Down Expand Up @@ -149,69 +150,93 @@ export class ExecuteWorkflow implements INodeType {
type: 'notice',
default: '',
},
{
displayName: 'Mode',
name: 'mode',
type: 'options',
noDataExpression: true,
options: [
{
// eslint-disable-next-line n8n-nodes-base/node-param-display-name-miscased
name: 'Run once with all items',
value: 'once',
description: 'Pass all items into a single execution of the sub-workflow',
},
{
// eslint-disable-next-line n8n-nodes-base/node-param-display-name-miscased
name: 'Run once for each item',
value: 'each',
description: 'Call the sub-workflow individually for each item',
},
],
default: 'once',
},
],
};

async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
const items = this.getInputData();
const source = this.getNodeParameter('source', 0) as string;
const mode = this.getNodeParameter('mode', 0, false) as string;
const items = this.getInputData();

const workflowInfo: IExecuteWorkflowInfo = {};

try {
if (source === 'database') {
// Read workflow from database
workflowInfo.id = this.getNodeParameter('workflowId', 0) as string;
} else if (source === 'localFile') {
// Read workflow from filesystem
const workflowPath = this.getNodeParameter('workflowPath', 0) as string;
if (mode === 'each') {
const returnData: INodeExecutionData[][] = [];

let workflowJson;
for (let i = 0; i < items.length; i++) {
try {
workflowJson = await fsReadFile(workflowPath, { encoding: 'utf8' });
const workflowInfo = await getWorkflowInfo.call(this, source, i);
const workflowResult: INodeExecutionData[][] = await this.executeWorkflow(workflowInfo, [
items[i],
]);

for (const [outputIndex, outputData] of workflowResult.entries()) {
for (const item of outputData) {
item.pairedItem = { item: i };
}

if (returnData[outputIndex] === undefined) {
returnData[outputIndex] = [];
}

returnData[outputIndex].push(...outputData);
}
} catch (error) {
if (error.code === 'ENOENT') {
throw new NodeOperationError(
this.getNode(),
`The file "${workflowPath}" could not be found.`,
);
if (this.continueOnFail()) {
return [[{ json: { error: error.message }, pairedItem: { item: i } }]];
}

throw error;
throw new NodeOperationError(this.getNode(), error, {
message: `Error executing workflow with item at index ${i}`,
description: error.message,
itemIndex: i,
});
}

workflowInfo.code = JSON.parse(workflowJson) as IWorkflowBase;
} else if (source === 'parameter') {
// Read workflow from parameter
const workflowJson = this.getNodeParameter('workflowJson', 0) as string;
workflowInfo.code = JSON.parse(workflowJson) as IWorkflowBase;
} else if (source === 'url') {
// Read workflow from url
const workflowUrl = this.getNodeParameter('workflowUrl', 0) as string;

const requestOptions = {
headers: {
accept: 'application/json,text/*;q=0.99',
},
method: 'GET',
uri: workflowUrl,
json: true,
gzip: true,
};

const response = await this.helpers.request(requestOptions);
workflowInfo.code = response;
}

const receivedData = await this.executeWorkflow(workflowInfo, items);
return returnData;
} else {
try {
const workflowInfo = await getWorkflowInfo.call(this, source);
const workflowResult: INodeExecutionData[][] = await this.executeWorkflow(
workflowInfo,
items,
);

return receivedData;
} catch (error) {
if (this.continueOnFail()) {
return [[{ json: { error: error.message } }]];
}
const pairedItem = generatePairedItemData(items.length);

throw error;
for (const output of workflowResult) {
for (const item of output) {
item.pairedItem = pairedItem;
}
}

return workflowResult;
} catch (error) {
const pairedItem = generatePairedItemData(items.length);
if (this.continueOnFail()) {
return [[{ json: { error: error.message }, pairedItem }]];
}
throw error;
}
}
}
}
58 changes: 58 additions & 0 deletions packages/nodes-base/nodes/ExecuteWorkflow/GenericFunctions.ts
@@ -0,0 +1,58 @@
import {
NodeOperationError,
type IExecuteFunctions,
type IExecuteWorkflowInfo,
jsonParse,
} from 'n8n-workflow';

import { readFile as fsReadFile } from 'fs/promises';

export async function getWorkflowInfo(this: IExecuteFunctions, source: string, itemIndex = 0) {
const workflowInfo: IExecuteWorkflowInfo = {};

if (source === 'database') {
// Read workflow from database
workflowInfo.id = this.getNodeParameter('workflowId', itemIndex) as string;
} else if (source === 'localFile') {
// Read workflow from filesystem
const workflowPath = this.getNodeParameter('workflowPath', itemIndex) as string;

let workflowJson;
try {
workflowJson = await fsReadFile(workflowPath, { encoding: 'utf8' });
} catch (error) {
if (error.code === 'ENOENT') {
throw new NodeOperationError(
this.getNode(),
`The file "${workflowPath}" could not be found, [item ${itemIndex}]`,
);
}

throw error;
}

workflowInfo.code = jsonParse(workflowJson);
} else if (source === 'parameter') {
// Read workflow from parameter
const workflowJson = this.getNodeParameter('workflowJson', itemIndex) as string;
workflowInfo.code = jsonParse(workflowJson);
} else if (source === 'url') {
// Read workflow from url
const workflowUrl = this.getNodeParameter('workflowUrl', itemIndex) as string;

const requestOptions = {
headers: {
accept: 'application/json,text/*;q=0.99',
},
method: 'GET',
uri: workflowUrl,
json: true,
gzip: true,
};

const response = await this.helpers.request(requestOptions);
workflowInfo.code = response;
}

return workflowInfo;
}
12 changes: 12 additions & 0 deletions packages/nodes-base/utils/utilities.ts
Expand Up @@ -3,6 +3,7 @@ import type {
IDisplayOptions,
INodeExecutionData,
INodeProperties,
IPairedItemData,
} from 'n8n-workflow';

import { jsonParse } from 'n8n-workflow';
Expand Down Expand Up @@ -291,3 +292,14 @@ export function flattenObject(data: IDataObject) {
}
return returnData;
}

/**
* Generate Paired Item Data by length of input array
*
* @param {number} length
*/
export function generatePairedItemData(length: number): IPairedItemData[] {
return Array.from({ length }, (_, item) => ({
item,
}));
}

0 comments on commit c8c14ca

Please sign in to comment.