mohiit1502/Automata
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(core): Add batching and other options to declarative nodes (#8885)

Browse Source 
Co-authored-by: Michael Kret <michael.k@radency.com>
This commit is contained in:
Jan Oberhauser
2024-06-06 22:39:31 -07:00
committed by GitHub
parent e520f8a98f
commit 4e568631be
4 changed files with 514 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

215
packages/workflow/src/RoutingNode.ts
View File

@@ -9,6 +9,7 @@
import get from 'lodash/get';
import merge from 'lodash/merge';
import set from 'lodash/set';
import url from 'node:url';
import type {
ICredentialDataDecryptedObject,
@@ -46,6 +47,7 @@ import type { Workflow } from './Workflow';
import { NodeOperationError } from './errors/node-operation.error';
import { NodeApiError } from './errors/node-api.error';
import { sleep } from './utils';
export class RoutingNode {
additionalData: IWorkflowExecuteAdditionalData;
@@ -76,6 +78,7 @@ export class RoutingNode {
this.workflow = workflow;
}
// eslint-disable-next-line complexity
async runNode(
inputData: ITaskDataConnections,
runIndex: number,
@@ -87,7 +90,6 @@ export class RoutingNode {
): Promise<INodeExecutionData[][] | null | undefined> {
const items = inputData.main[0] as INodeExecutionData[];
const returnData: INodeExecutionData[] = [];
let responseData;
let credentialType: string | undefined;
@@ -129,24 +131,41 @@ export class RoutingNode {
}
}
// TODO: Think about how batching could be handled for REST APIs which support it
for (let i = 0; i < items.length; i++) {
let thisArgs: IExecuteSingleFunctions | undefined;
try {
thisArgs = nodeExecuteFunctions.getExecuteSingleFunctions(
const { batching } = executeFunctions.getNodeParameter('requestOptions', 0, {}) as {
batching: { batch: { batchSize: number; batchInterval: number } };
};
const batchSize = batching?.batch?.batchSize > 0 ? batching?.batch?.batchSize : 1;
const batchInterval = batching?.batch.batchInterval;
const requestPromises = [];
const itemContext: Array<{
thisArgs: IExecuteSingleFunctions;
requestData: DeclarativeRestApiSettings.ResultOptions;
}> = [];
for (let itemIndex = 0; itemIndex < items.length; itemIndex++) {
if (itemIndex > 0 && batchSize >= 0 && batchInterval > 0) {
if (itemIndex % batchSize === 0) {
await sleep(batchInterval);
}
}
itemContext.push({
thisArgs: nodeExecuteFunctions.getExecuteSingleFunctions(
this.workflow,
this.runExecutionData,
runIndex,
this.connectionInputData,
inputData,
this.node,
i,
itemIndex,
this.additionalData,
executeData,
this.mode,
abortSignal,
);
const requestData: DeclarativeRestApiSettings.ResultOptions = {
),
requestData: {
options: {
qs: {},
body: {},
@@ -155,88 +174,160 @@ export class RoutingNode {
preSend: [],
postReceive: [],
requestOperations: {},
} as DeclarativeRestApiSettings.ResultOptions,
});
const { proxy, timeout, allowUnauthorizedCerts } = itemContext[
itemIndex
].thisArgs.getNodeParameter('requestOptions', 0, {}) as {
proxy: string;
timeout: number;
allowUnauthorizedCerts: boolean;
};
if (nodeType.description.requestOperations) {
itemContext[itemIndex].requestData.requestOperations = {
...nodeType.description.requestOperations,
};
}
if (nodeType.description.requestOperations) {
requestData.requestOperations = { ...nodeType.description.requestOperations };
}
if (nodeType.description.requestDefaults) {
for (const key of Object.keys(nodeType.description.requestDefaults)) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let value = (nodeType.description.requestDefaults as Record<string, any>)[key];
// If the value is an expression resolve it
value = this.getParameterValue(
value,
i,
runIndex,
executeData,
{ $credentials: credentials, $version: this.node.typeVersion },
false,
) as string;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(requestData.options as Record<string, any>)[key] = value;
}
}
for (const property of nodeType.description.properties) {
let value = get(this.node.parameters, property.name, []) as string | NodeParameterValue;
if (nodeType.description.requestDefaults) {
for (const key of Object.keys(nodeType.description.requestDefaults)) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let value = (nodeType.description.requestDefaults as Record<string, any>)[key];
// If the value is an expression resolve it
value = this.getParameterValue(
value,
i,
itemIndex,
runIndex,
executeData,
{ $credentials: credentials, $version: this.node.typeVersion },
false,
) as string | NodeParameterValue;
const tempOptions = this.getRequestOptionsFromParameters(
thisArgs,
property,
i,
runIndex,
'',
{ $credentials: credentials, $value: value, $version: this.node.typeVersion },
);
this.mergeOptions(requestData, tempOptions);
) as string;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(itemContext[itemIndex].requestData.options as Record<string, any>)[key] = value;
}
}
// TODO: Change to handle some requests in parallel (should be configurable)
responseData = await this.makeRoutingRequest(
requestData,
thisArgs,
i,
for (const property of nodeType.description.properties) {
let value = get(this.node.parameters, property.name, []) as string | NodeParameterValue;
// If the value is an expression resolve it
value = this.getParameterValue(
value,
itemIndex,
runIndex,
credentialType,
requestData.requestOperations,
credentialsDecrypted,
executeData,
{ $credentials: credentials, $version: this.node.typeVersion },
false,
) as string | NodeParameterValue;
const tempOptions = this.getRequestOptionsFromParameters(
itemContext[itemIndex].thisArgs,
property,
itemIndex,
runIndex,
'',
{ $credentials: credentials, $value: value, $version: this.node.typeVersion },
);
if (requestData.maxResults) {
// Remove not needed items in case APIs return to many
responseData.splice(requestData.maxResults as number);
this.mergeOptions(itemContext[itemIndex].requestData, tempOptions);
}
if (proxy) {
const proxyParsed = url.parse(proxy);
const proxyProperties = ['host', 'port'];
for (const property of proxyProperties) {
if (
!(property in proxyParsed) ||
proxyParsed[property as keyof typeof proxyParsed] === null
) {
throw new NodeOperationError(this.node, 'The proxy is not value', {
runIndex,
itemIndex,
description: `The proxy URL does not contain a valid value for "${property}"`,
});
}
}
returnData.push(...responseData);
} catch (error) {
if (thisArgs !== undefined && thisArgs.continueOnFail()) {
itemContext[itemIndex].requestData.options.proxy = {
host: proxyParsed.hostname as string,
port: parseInt(proxyParsed.port!),
protocol: proxyParsed.protocol?.replace(/:$/, '') || undefined,
};
if (proxyParsed.auth) {
const [username, password] = proxyParsed.auth.split(':');
itemContext[itemIndex].requestData.options.proxy!.auth = {
username,
password,
};
}
}
if (allowUnauthorizedCerts) {
itemContext[itemIndex].requestData.options.skipSslCertificateValidation =
allowUnauthorizedCerts;
}
if (timeout) {
itemContext[itemIndex].requestData.options.timeout = timeout;
} else {
// set default timeout to 5 minutes
itemContext[itemIndex].requestData.options.timeout = 300_000;
}
requestPromises.push(
this.makeRoutingRequest(
itemContext[itemIndex].requestData,
itemContext[itemIndex].thisArgs,
itemIndex,
runIndex,
credentialType,
itemContext[itemIndex].requestData.requestOperations,
credentialsDecrypted,
),
);
}
const promisesResponses = await Promise.allSettled(requestPromises);
let responseData: any;
for (let itemIndex = 0; itemIndex < items.length; itemIndex++) {
responseData = promisesResponses.shift();
if (responseData!.status !== 'fulfilled') {
if (responseData.reason.statusCode === 429) {
responseData.reason.message =
"Try spacing your requests out using the batching settings under 'Options'";
}
const error = responseData.reason;
if (itemContext[itemIndex].thisArgs?.continueOnFail()) {
returnData.push({ json: {}, error: error as NodeApiError });
continue;
}
if (error instanceof NodeApiError) {
set(error, 'context.itemIndex', i);
set(error, 'context.itemIndex', itemIndex);
set(error, 'context.runIndex', runIndex);
throw error;
}
throw new NodeApiError(this.node, error as JsonObject, {
runIndex,
itemIndex: i,
itemIndex,
message: error?.message,
description: error?.description,
httpCode: error.isAxiosError && error.response ? String(error.response?.status) : 'none',
});
}
if (itemContext[itemIndex].requestData.maxResults) {
// Remove not needed items in case APIs return to many
responseData.value.splice(itemContext[itemIndex].requestData.maxResults as number);
}
returnData.push(...responseData.value);
}
return [returnData];