From 0ccf47044a2ba5b94140bfdd2ba36b868091288d Mon Sep 17 00:00:00 2001 From: yehorkardash Date: Wed, 5 Nov 2025 10:57:08 +0200 Subject: [PATCH] feat(Extract from File Node): Add `Skip Records With Errors` option (#21347) Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> Co-authored-by: Michael Kret --- .../ExtractFromFile/ExtractFromFile.node.ts | 7 +- .../actions/spreadsheet.operation.ts | 14 +- .../nodes/SpreadsheetFile/description.ts | 34 +++++ .../test/fromFile.operation.test.ts | 122 +++++++++++++++++- .../SpreadsheetFile/v2/fromFile.operation.ts | 72 +++++++++-- 5 files changed, 233 insertions(+), 16 deletions(-) diff --git a/packages/nodes-base/nodes/Files/ExtractFromFile/ExtractFromFile.node.ts b/packages/nodes-base/nodes/Files/ExtractFromFile/ExtractFromFile.node.ts index 81408a6cca5..7f68db3f330 100644 --- a/packages/nodes-base/nodes/Files/ExtractFromFile/ExtractFromFile.node.ts +++ b/packages/nodes-base/nodes/Files/ExtractFromFile/ExtractFromFile.node.ts @@ -17,7 +17,7 @@ export class ExtractFromFile implements INodeType { name: 'extractFromFile', icon: { light: 'file:extractFromFile.svg', dark: 'file:extractFromFile.dark.svg' }, group: ['input'], - version: 1, + version: [1, 1.1], description: 'Convert binary data to JSON', defaults: { name: 'Extract from File', @@ -114,12 +114,15 @@ export class ExtractFromFile implements INodeType { }; async execute(this: IExecuteFunctions) { + const version = this.getNode().typeVersion; const items = this.getInputData(); const operation = this.getNodeParameter('operation', 0); let returnData: INodeExecutionData[] = []; if (spreadsheet.operations.includes(operation)) { - returnData = await spreadsheet.execute.call(this, items, 'operation'); + returnData = await spreadsheet.execute.call(this, items, 'operation', { + failOnCsvBufferError: version > 1, + }); } if (['binaryToPropery', 'fromJson', 'text', 'fromIcs', 'xml'].includes(operation)) { diff --git a/packages/nodes-base/nodes/Files/ExtractFromFile/actions/spreadsheet.operation.ts b/packages/nodes-base/nodes/Files/ExtractFromFile/actions/spreadsheet.operation.ts index 9c3d98b3478..6b5e1485e1a 100644 --- a/packages/nodes-base/nodes/Files/ExtractFromFile/actions/spreadsheet.operation.ts +++ b/packages/nodes-base/nodes/Files/ExtractFromFile/actions/spreadsheet.operation.ts @@ -18,9 +18,15 @@ export const description: INodeProperties[] = fromFile.description newProperty.options = (newProperty.options as INodeProperties[]).map((option) => { let newOption = option; if ( - ['delimiter', 'encoding', 'fromLine', 'maxRowCount', 'enableBOM', 'relaxQuotes'].includes( - option.name, - ) + [ + 'delimiter', + 'encoding', + 'fromLine', + 'maxRowCount', + 'enableBOM', + 'relaxQuotes', + 'skipRecordsWithErrors', + ].includes(option.name) ) { newOption = { ...option, displayOptions: { show: { '/operation': ['csv'] } } }; } @@ -53,11 +59,13 @@ export async function execute( this: IExecuteFunctions, items: INodeExecutionData[], fileFormatProperty: string, + options?: fromFile.FromFileOptions, ) { const returnData: INodeExecutionData[] = await fromFile.execute.call( this, items, fileFormatProperty, + options, ); return returnData; } diff --git a/packages/nodes-base/nodes/SpreadsheetFile/description.ts b/packages/nodes-base/nodes/SpreadsheetFile/description.ts index 2def6d5c91c..af25efe15ad 100644 --- a/packages/nodes-base/nodes/SpreadsheetFile/description.ts +++ b/packages/nodes-base/nodes/SpreadsheetFile/description.ts @@ -298,5 +298,39 @@ export const fromFileOptions: INodeProperties = { placeholder: 'e.g. 0', description: 'Start handling records from the requested line number. Starts at 0.', }, + { + displayName: 'Skip Records With Errors', + name: 'skipRecordsWithErrors', + type: 'fixedCollection', + default: { value: { enabled: true, maxSkippedRecords: -1 } }, + options: [ + { + displayName: 'Value', + name: 'value', + values: [ + { + displayName: 'Enabled', + name: 'enabled', + type: 'boolean', + default: false, + description: 'Whether to skip records with errors when reading from file', + }, + { + displayName: 'Max Skipped Records', + name: 'maxSkippedRecords', + type: 'number', + default: -1, + description: + 'The maximum number of records that can be skipped, will throw an error if exceeded. Set to -1 to remove limit.', + }, + ], + }, + ], + displayOptions: { + show: { + '/fileFormat': ['csv'], + }, + }, + }, ], }; diff --git a/packages/nodes-base/nodes/SpreadsheetFile/test/fromFile.operation.test.ts b/packages/nodes-base/nodes/SpreadsheetFile/test/fromFile.operation.test.ts index 3be7fe4b32e..0c1d71e67de 100644 --- a/packages/nodes-base/nodes/SpreadsheetFile/test/fromFile.operation.test.ts +++ b/packages/nodes-base/nodes/SpreadsheetFile/test/fromFile.operation.test.ts @@ -1,5 +1,5 @@ import { mockDeep } from 'jest-mock-extended'; -import type { IExecuteFunctions, INodeExecutionData, IBinaryData, INode } from 'n8n-workflow'; +import type { IBinaryData, IExecuteFunctions, INode, INodeExecutionData } from 'n8n-workflow'; import { BINARY_ENCODING, NodeOperationError } from 'n8n-workflow'; import { Readable } from 'stream'; @@ -677,4 +677,124 @@ describe('fromFile.operation - xlsx parsing logic', () => { }); }); }); + + describe('CSV parsing with skipRecordsWithErrors', () => { + const invalidCsvData = 'id,name\n3,"John"\n1,"Alice\n2,"Bob"'; + const mockBinaryDataCSV: IBinaryData = { + data: Buffer.from(invalidCsvData, 'utf8').toString(BINARY_ENCODING), + mimeType: 'text/csv', + fileExtension: 'csv', + fileName: 'test.csv', + }; + + beforeEach(() => { + jest.clearAllMocks(); + mockExecuteFunctions.getNodeParameter.mockImplementation( + (paramName: string, _itemIndex: number, defaultValue?: any) => { + switch (paramName) { + case 'fileFormat': + return 'csv'; + case 'binaryPropertyName': + return 'data'; + case 'options': + return {}; + default: + return defaultValue; + } + }, + ); + mockExecuteFunctions.helpers.assertBinaryData.mockReturnValue(mockBinaryDataCSV); + mockExecuteFunctions.getNode.mockReturnValue({ + name: 'SpreadsheetFile', + type: 'n8n-nodes-base.spreadsheetFile', + id: 'test-node-id', + } as INode); + mockExecuteFunctions.continueOnFail.mockReturnValue(false); + }); + + it('should skip records with errors when skipRecordsWithErrors is enabled with limit -1', async () => { + mockExecuteFunctions.getNodeParameter.mockImplementation((paramName: string) => { + if (paramName === 'fileFormat') return 'csv'; + if (paramName === 'binaryPropertyName') return 'data'; + if (paramName === 'options') + return { + skipRecordsWithErrors: { value: { enabled: true, maxSkippedRecords: -1 } }, + columns: true, + }; + return undefined; + }); + + const items: INodeExecutionData[] = [{ json: {} }]; + + const result = await execute.call(mockExecuteFunctions, items); + + // Should have 1 valid record (John), Bob and Alice is considered a single record with error + expect(result).toHaveLength(1); + expect(result[0].json).toEqual({ id: '3', name: 'John' }); + }); + + it('should skip records with errors when skipRecordsWithErrors is enabled with limit 1', async () => { + mockExecuteFunctions.getNodeParameter.mockImplementation((paramName: string) => { + if (paramName === 'fileFormat') return 'csv'; + if (paramName === 'binaryPropertyName') return 'data'; + if (paramName === 'options') + return { + skipRecordsWithErrors: { value: { enabled: true, maxSkippedRecords: 1 } }, + columns: true, + }; + return undefined; + }); + + const items: INodeExecutionData[] = [{ json: {} }]; + + const result = await execute.call(mockExecuteFunctions, items); + + expect(result).toHaveLength(1); + expect(result[0].json).toEqual({ id: '3', name: 'John' }); + expect(result[0].pairedItem).toEqual({ item: 0 }); + }); + + it('should throw error when skipped records exceed maxSkippedRecords limit', async () => { + mockExecuteFunctions.getNodeParameter.mockImplementation((paramName: string) => { + if (paramName === 'fileFormat') return 'csv'; + if (paramName === 'binaryPropertyName') return 'data'; + if (paramName === 'options') + return { + skipRecordsWithErrors: { value: { enabled: true, maxSkippedRecords: 1 } }, + columns: true, + }; + return undefined; + }); + + const csvWithThreeErrors = + 'id,name\n3,"John"\n1,"Alice\n2,"Bob"\n4,"Charlie\n5,"Eve\n6,"David'; + const mockBinaryDataThreeErrors: IBinaryData = { + data: Buffer.from(csvWithThreeErrors, 'utf8').toString(BINARY_ENCODING), + mimeType: 'text/csv', + fileExtension: 'csv', + fileName: 'test-three-errors.csv', + }; + + mockExecuteFunctions.helpers.assertBinaryData.mockReturnValue(mockBinaryDataThreeErrors); + + const items: INodeExecutionData[] = [{ json: {} }]; + + mockExecuteFunctions.getNodeParameter.mockImplementation((paramName: string) => { + if (paramName === 'fileFormat') return 'csv'; + if (paramName === 'binaryPropertyName') return 'data'; + if (paramName === 'options') + return { + skipRecordsWithErrors: { value: { enabled: true, maxSkippedRecords: 1 } }, + columns: true, + }; + return undefined; + }); + + mockExecuteFunctions.helpers.assertBinaryData.mockReturnValue(mockBinaryDataThreeErrors); + + await expect(execute.call(mockExecuteFunctions, items)).rejects.toThrow( + 'Max number of skipped records exceeded', + ); + }); + }); }); diff --git a/packages/nodes-base/nodes/SpreadsheetFile/v2/fromFile.operation.ts b/packages/nodes-base/nodes/SpreadsheetFile/v2/fromFile.operation.ts index 3e4d42ff935..59a956620da 100644 --- a/packages/nodes-base/nodes/SpreadsheetFile/v2/fromFile.operation.ts +++ b/packages/nodes-base/nodes/SpreadsheetFile/v2/fromFile.operation.ts @@ -1,4 +1,4 @@ -import { parse as createCSVParser } from 'csv-parse'; +import { parse as createCSVParser, type Options as CSVOptions } from 'csv-parse'; import type { IExecuteFunctions, INodeExecutionData, INodeProperties } from 'n8n-workflow'; import { BINARY_ENCODING, NodeOperationError } from 'n8n-workflow'; import type { Sheet2JSONOpts, ParsingOptions } from 'xlsx'; @@ -6,6 +6,25 @@ import { read as xlsxRead, utils as xlsxUtils } from 'xlsx'; import { binaryProperty, fromFileOptions } from '../description'; +interface Options { + maxRowCount?: number; + delimiter?: string; + fromLine?: number; + encoding?: BufferEncoding; + enableBOM?: boolean; + skipRecordsWithErrors?: { + value?: { enabled?: boolean; maxSkippedRecords?: number }; + }; + to?: number; + relaxQuotes?: boolean; + includeEmptyCells?: boolean; + rawData?: boolean; + readAsString?: boolean; + sheetName?: string; + range?: number | string; + headerRow?: boolean; +} + export const description: INodeProperties[] = [ binaryProperty, { @@ -59,10 +78,15 @@ export const description: INodeProperties[] = [ fromFileOptions, ]; +export interface FromFileOptions { + failOnCsvBufferError?: boolean; +} + export async function execute( this: IExecuteFunctions, items: INodeExecutionData[], fileFormatProperty = 'fileFormat', + { failOnCsvBufferError = false }: FromFileOptions = {}, ) { const returnData: INodeExecutionData[] = []; let fileExtension; @@ -70,7 +94,7 @@ export async function execute( for (let i = 0; i < items.length; i++) { try { - const options = this.getNodeParameter('options', i, {}); + const options = this.getNodeParameter('options', i, {}) as Options; fileFormat = this.getNodeParameter(fileFormatProperty, i, ''); const binaryPropertyName = this.getNodeParameter('binaryPropertyName', i); const binaryData = this.helpers.assertBinaryData(i, binaryPropertyName); @@ -88,14 +112,16 @@ export async function execute( if (fileFormat === 'csv') { const maxRowCount = options.maxRowCount as number; - const parser = createCSVParser({ - delimiter: options.delimiter as string, - fromLine: options.fromLine as number, - encoding: options.encoding as BufferEncoding, - bom: options.enableBOM as boolean, + const skipRecordsWithErrors = options.skipRecordsWithErrors?.value?.enabled; + const csvOptions: CSVOptions = { + delimiter: options.delimiter, + fromLine: options.fromLine, + encoding: options.encoding, + bom: options.enableBOM, to: maxRowCount > -1 ? maxRowCount : undefined, + skip_records_with_error: skipRecordsWithErrors, columns: options.headerRow !== false, - relax_quotes: options.relaxQuotes as boolean, + relax_quotes: options.relaxQuotes, onRecord: (record) => { if (!options.includeEmptyCells) { record = Object.fromEntries( @@ -104,10 +130,17 @@ export async function execute( } rows.push(record); }, + }; + const parser = createCSVParser(csvOptions); + + let skippedRecords = 0; + parser.on('skip', (_err) => { + skippedRecords += 1; }); + if (binaryData.id) { const stream = await this.helpers.getBinaryStream(binaryData.id); - await new Promise(async (resolve, reject) => { + await new Promise((resolve, reject) => { parser.on('error', reject); parser.on('readable', () => { stream.unpipe(parser); @@ -118,7 +151,26 @@ export async function execute( }); } else { parser.write(binaryData.data, BINARY_ENCODING); - parser.end(); + + if (failOnCsvBufferError) { + await new Promise((resolve, reject) => { + parser.on('error', reject); + parser.on('readable', () => { + resolve(); + }); + parser.end(); + }); + } else { + // this ignores errors, but we keep it for backwards compatibility + parser.end(); + } + } + + const maxSkippedRecords = options.skipRecordsWithErrors?.value?.maxSkippedRecords ?? -1; + if (skipRecordsWithErrors && maxSkippedRecords > 0 && skippedRecords > maxSkippedRecords) { + throw new NodeOperationError(this.getNode(), 'Max number of skipped records exceeded', { + itemIndex: i, + }); } } else { const xlsxOptions: ParsingOptions = { raw: options.rawData as boolean };