✨ MSSQL Node Integration (#729)

* basic setup mssql node

* executeQuery for MSSQL working

* work on insert MSSQL, incomplete

* 🚧 basic setup update functionality

* 🔨 refactor insert for handling >1000 values

* ✨ complete MSSQL node

* ✨ add delete action to node

* 🚧 handling multiple tables and column sets

* 🐛 enabling usage of expression for every field

* 🔨 remove lodash dependency

* ✨ enable continue on fail option

* 🐎 minify icon

* 🔨 improve table creation, item copying, 🐛 correct output of node when active continue on fail

* 🐛 move mssql types to dev dependencies

* 🎨 remove auto formatting from redis

* 🎨 apply corrected syntax format

* ⏪ reset Redis node to master stage

* 🐛 fix building issue

packages/nodes-base/credentials/MicrosoftSqlServer.credentials.ts

@@ -0,0 +1,47 @@
+import { ICredentialType, NodePropertyTypes } from 'n8n-workflow';
+
+export class MicrosoftSqlServer implements ICredentialType {
+	name = 'microsoftSqlServer';
+	displayName = 'Microsoft SQL Server';
+	properties = [
+		{
+			displayName: 'Server',
+			name: 'server',
+			type: 'string' as NodePropertyTypes,
+			default: 'localhost'
+		},
+		{
+			displayName: 'Database',
+			name: 'database',
+			type: 'string' as NodePropertyTypes,
+			default: 'master'
+		},
+		{
+			displayName: 'User',
+			name: 'user',
+			type: 'string' as NodePropertyTypes,
+			default: 'sa'
+		},
+		{
+			displayName: 'Password',
+			name: 'password',
+			type: 'string' as NodePropertyTypes,
+			typeOptions: {
+				password: true
+			},
+			default: ''
+		},
+		{
+			displayName: 'Port',
+			name: 'port',
+			type: 'number' as NodePropertyTypes,
+			default: 1433
+		},
+		{
+			displayName: 'Domain',
+			name: 'domain',
+			type: 'string' as NodePropertyTypes,
+			default: ''
+		}
+	];
+}

packages/nodes-base/nodes/Microsoft/SqlServer/GenericFunctions.ts

@@ -0,0 +1,144 @@
+import { IDataObject, INodeExecutionData } from 'n8n-workflow';
+import { ITables } from './TableInterface';
+
+/**
+ * Returns a copy of the item which only contains the json data and
+ * of that only the defined properties
+ *
+ * @param {INodeExecutionData} item The item to copy
+ * @param {string[]} properties The properties it should include
+ * @returns
+ */
+export function copyInputItem(
+	item: INodeExecutionData,
+	properties: string[],
+): IDataObject {
+	// Prepare the data to insert and copy it to be returned
+	let newItem: IDataObject = {};
+	for (const property of properties) {
+		if (item.json[property] === undefined) {
+			newItem[property] = null;
+		} else {
+			newItem[property] = JSON.parse(JSON.stringify(item.json[property]));
+		}
+	}
+	return newItem;
+}
+
+/**
+ * Creates an ITables with the columns for the operations
+ *
+ * @param {INodeExecutionData[]} items The items to extract the tables/columns for
+ * @param {function} getNodeParam getter for the Node's Parameters
+ * @returns {ITables} {tableName: {colNames: [items]}};
+ */
+export function createTableStruct(
+	getNodeParam: Function,
+	items: INodeExecutionData[],
+	additionalProperties: string[] = [],
+	keyName?: string,
+): ITables {
+	return items.reduce((tables, item, index) => {
+		const table = getNodeParam('table', index) as string;
+		const columnString = getNodeParam('columns', index) as string;
+		const columns = columnString.split(',').map(column => column.trim());
+		const itemCopy = copyInputItem(item, columns.concat(additionalProperties));
+		const keyParam = keyName
+			? (getNodeParam(keyName, index) as string)
+			: undefined;
+		if (tables[table] === undefined) {
+			tables[table] = {};
+		}
+		if (tables[table][columnString] === undefined) {
+			tables[table][columnString] = [];
+		}
+		if (keyName) {
+			itemCopy[keyName] = keyParam;
+		}
+		tables[table][columnString].push(itemCopy);
+		return tables;
+	}, {} as ITables);
+}
+
+/**
+ * Executes a queue of queries on given ITables.
+ *
+ * @param {ITables} tables The ITables to be processed.
+ * @param {function} buildQueryQueue function that builds the queue of promises
+ * @returns {Promise}
+ */
+export function executeQueryQueue(
+	tables: ITables,
+	buildQueryQueue: Function,
+): Promise<any[]> {
+	return Promise.all(
+		Object.keys(tables).map(table => {
+			const columnsResults = Object.keys(tables[table]).map(columnString => {
+				return Promise.all(
+					buildQueryQueue({
+						table: table,
+						columnString: columnString,
+						items: tables[table][columnString],
+					}),
+				);
+			});
+			return Promise.all(columnsResults);
+		}),
+	);
+}
+
+/**
+ * Extracts the values from the item for INSERT
+ *
+ * @param {IDataObject} item The item to extract
+ * @returns {string} (Val1, Val2, ...)
+ */
+export function extractValues(item: IDataObject): string {
+	return `(${Object.values(item as any)
+		.map(val => (typeof val === 'string' ? `'${val}'` : val)) // maybe other types such as dates have to be handled as well
+		.join(',')})`;
+}
+
+/**
+ * Extracts the SET from the item for UPDATE
+ *
+ * @param {IDataObject} item The item to extract from
+ * @param {string[]} columns The columns to update
+ * @returns {string} col1 = val1, col2 = val2
+ */
+export function extractUpdateSet(item: IDataObject, columns: string[]): string {
+	return columns
+		.map(
+			column =>
+				`${column} = ${
+					typeof item[column] === 'string' ? `'${item[column]}'` : item[column]
+				}`,
+		)
+		.join(',');
+}
+
+/**
+ * Extracts the WHERE condition from the item for UPDATE
+ *
+ * @param {IDataObject} item The item to extract from
+ * @param {string} key The column name to build the condition with
+ * @returns {string} id = '123'
+ */
+export function extractUpdateCondition(item: IDataObject, key: string): string {
+	return `${key} = ${
+		typeof item[key] === 'string' ? `'${item[key]}'` : item[key]
+	}`;
+}
+
+/**
+ * Extracts the WHERE condition from the items for DELETE
+ *
+ * @param {IDataObject[]} items The items to extract the values from
+ * @param {string} key The column name to extract the value from for the delete condition
+ * @returns {string} (Val1, Val2, ...)
+ */
+export function extractDeleteValues(items: IDataObject[], key: string): string {
+	return `(${items
+		.map(item => (typeof item[key] === 'string' ? `'${item[key]}'` : item[key]))
+		.join(',')})`;
+}

packages/nodes-base/nodes/Microsoft/SqlServer/MicrosoftSqlServer.node.ts

@@ -0,0 +1,410 @@
+import { IExecuteFunctions } from 'n8n-core';
+import {
+	IDataObject,
+	INodeExecutionData,
+	INodeType,
+	INodeTypeDescription,
+} from 'n8n-workflow';
+
+import { chunk, flatten } from '../../utils/utilities';
+
+import * as mssql from 'mssql';
+
+import { ITables } from './TableInterface';
+
+import {
+	copyInputItem,
+	createTableStruct,
+	executeQueryQueue,
+	extractDeleteValues,
+	extractUpdateCondition,
+	extractUpdateSet,
+	extractValues,
+} from './GenericFunctions';
+
+export class MicrosoftSqlServer implements INodeType {
+	description: INodeTypeDescription = {
+		displayName: 'Microsoft SQL Server',
+		name: 'microsoftSqlServer',
+		icon: 'file:mssql.png',
+		group: ['input'],
+		version: 1,
+		description: 'Gets, add and update data in Microsoft SQL Server.',
+		defaults: {
+			name: 'Microsoft SQL Server',
+			color: '#1d4bab',
+		},
+		inputs: ['main'],
+		outputs: ['main'],
+		credentials: [
+			{
+				name: 'microsoftSqlServer',
+				required: true,
+			},
+		],
+		properties: [
+			{
+				displayName: 'Operation',
+				name: 'operation',
+				type: 'options',
+				options: [
+					{
+						name: 'Execute Query',
+						value: 'executeQuery',
+						description: 'Executes a SQL query.',
+					},
+					{
+						name: 'Insert',
+						value: 'insert',
+						description: 'Insert rows in database.',
+					},
+					{
+						name: 'Update',
+						value: 'update',
+						description: 'Updates rows in database.',
+					},
+					{
+						name: 'Delete',
+						value: 'delete',
+						description: 'Deletes rows in database.',
+					},
+				],
+				default: 'insert',
+				description: 'The operation to perform.',
+			},
+
+			// ----------------------------------
+			//         executeQuery
+			// ----------------------------------
+			{
+				displayName: 'Query',
+				name: 'query',
+				type: 'string',
+				typeOptions: {
+					rows: 5,
+				},
+				displayOptions: {
+					show: {
+						operation: ['executeQuery'],
+					},
+				},
+				default: '',
+				placeholder: 'SELECT id, name FROM product WHERE id < 40',
+				required: true,
+				description: 'The SQL query to execute.',
+			},
+
+			// ----------------------------------
+			//         insert
+			// ----------------------------------
+			{
+				displayName: 'Table',
+				name: 'table',
+				type: 'string',
+				displayOptions: {
+					show: {
+						operation: ['insert'],
+					},
+				},
+				default: '',
+				required: true,
+				description: 'Name of the table in which to insert data to.',
+			},
+			{
+				displayName: 'Columns',
+				name: 'columns',
+				type: 'string',
+				displayOptions: {
+					show: {
+						operation: ['insert'],
+					},
+				},
+				default: '',
+				placeholder: 'id,name,description',
+				description:
+					'Comma separated list of the properties which should used as columns for the new rows.',
+			},
+
+			// ----------------------------------
+			//         update
+			// ----------------------------------
+			{
+				displayName: 'Table',
+				name: 'table',
+				type: 'string',
+				displayOptions: {
+					show: {
+						operation: ['update'],
+					},
+				},
+				default: '',
+				required: true,
+				description: 'Name of the table in which to update data in',
+			},
+			{
+				displayName: 'Update Key',
+				name: 'updateKey',
+				type: 'string',
+				displayOptions: {
+					show: {
+						operation: ['update'],
+					},
+				},
+				default: 'id',
+				required: true,
+				description:
+					'Name of the property which decides which rows in the database should be updated. Normally that would be "id".',
+			},
+			{
+				displayName: 'Columns',
+				name: 'columns',
+				type: 'string',
+				displayOptions: {
+					show: {
+						operation: ['update'],
+					},
+				},
+				default: '',
+				placeholder: 'name,description',
+				description:
+					'Comma separated list of the properties which should used as columns for rows to update.',
+			},
+
+			// ----------------------------------
+			//         delete
+			// ----------------------------------
+			{
+				displayName: 'Table',
+				name: 'table',
+				type: 'string',
+				displayOptions: {
+					show: {
+						operation: ['delete'],
+					},
+				},
+				default: '',
+				required: true,
+				description: 'Name of the table in which to delete data.',
+			},
+			{
+				displayName: 'Delete Key',
+				name: 'deleteKey',
+				type: 'string',
+				displayOptions: {
+					show: {
+						operation: ['delete'],
+					},
+				},
+				default: 'id',
+				required: true,
+				description:
+					'Name of the property which decides which rows in the database should be deleted. Normally that would be "id".',
+			},
+		],
+	};
+
+	async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
+		const credentials = this.getCredentials('microsoftSqlServer');
+
+		if (credentials === undefined) {
+			throw new Error('No credentials got returned!');
+		}
+
+		const config = {
+			server: credentials.server as string,
+			port: credentials.port as number,
+			database: credentials.database as string,
+			user: credentials.user as string,
+			password: credentials.password as string,
+			domain: credentials.domain ? (credentials.domain as string) : undefined,
+		};
+
+		const pool = new mssql.ConnectionPool(config);
+		await pool.connect();
+
+		let returnItems: any = [];
+
+		const items = this.getInputData();
+		const operation = this.getNodeParameter('operation', 0) as string;
+
+		try {
+			if (operation === 'executeQuery') {
+				// ----------------------------------
+				//         executeQuery
+				// ----------------------------------
+
+				const rawQuery = this.getNodeParameter('query', 0) as string;
+
+				const queryResult = await pool.request().query(rawQuery);
+
+				const result =
+					queryResult.recordsets.length > 1
+						? flatten(queryResult.recordsets)
+						: queryResult.recordsets[0];
+
+				returnItems = this.helpers.returnJsonArray(result as IDataObject[]);
+			} else if (operation === 'insert') {
+				// ----------------------------------
+				//         insert
+				// ----------------------------------
+
+				const tables = createTableStruct(this.getNodeParameter, items);
+				const queriesResults = await executeQueryQueue(
+					tables,
+					({
+						table,
+						columnString,
+						items,
+					}: {
+						table: string;
+						columnString: string;
+						items: IDataObject[];
+					}): Promise<any>[] => {
+						return chunk(items, 1000).map(insertValues => {
+							const values = insertValues
+								.map((item: IDataObject) => extractValues(item))
+								.join(',');
+
+							return pool
+								.request()
+								.query(
+									`INSERT INTO ${table}(${columnString}) VALUES ${values};`,
+								);
+						});
+					},
+				);
+
+				const rowsAffected = flatten(queriesResults).reduce(
+					(acc: number, resp: mssql.IResult<object>): number =>
+						(acc += resp.rowsAffected.reduce((sum, val) => (sum += val))),
+					0,
+				);
+
+				returnItems = this.helpers.returnJsonArray({
+					rowsAffected,
+				} as IDataObject);
+			} else if (operation === 'update') {
+				// ----------------------------------
+				//         update
+				// ----------------------------------
+
+				const updateKeys = items.map(
+					(item, index) => this.getNodeParameter('updateKey', index) as string,
+				);
+				const tables = createTableStruct(
+					this.getNodeParameter,
+					items,
+					['updateKey'],
+					'updateKey',
+				);
+				const queriesResults = await executeQueryQueue(
+					tables,
+					({
+						table,
+						columnString,
+						items,
+					}: {
+						table: string;
+						columnString: string;
+						items: IDataObject[];
+					}): Promise<any>[] => {
+						return items.map(item => {
+							const columns = columnString
+								.split(',')
+								.map(column => column.trim());
+
+							const setValues = extractUpdateSet(item, columns);
+							const condition = extractUpdateCondition(
+								item,
+								item.updateKey as string,
+							);
+
+							return pool
+								.request()
+								.query(`UPDATE ${table} SET ${setValues} WHERE ${condition};`);
+						});
+					},
+				);
+
+				const rowsAffected = flatten(queriesResults).reduce(
+					(acc: number, resp: mssql.IResult<object>): number =>
+						(acc += resp.rowsAffected.reduce((sum, val) => (sum += val))),
+					0,
+				);
+
+				returnItems = this.helpers.returnJsonArray({
+					rowsAffected,
+				} as IDataObject);
+			} else if (operation === 'delete') {
+				// ----------------------------------
+				//         delete
+				// ----------------------------------
+
+				const tables = items.reduce((tables, item, index) => {
+					const table = this.getNodeParameter('table', index) as string;
+					const deleteKey = this.getNodeParameter('deleteKey', index) as string;
+					if (tables[table] === undefined) {
+						tables[table] = {};
+					}
+					if (tables[table][deleteKey] === undefined) {
+						tables[table][deleteKey] = [];
+					}
+					tables[table][deleteKey].push(item);
+					return tables;
+				}, {} as ITables);
+
+				const queriesResults = await Promise.all(
+					Object.keys(tables).map(table => {
+						const deleteKeyResults = Object.keys(tables[table]).map(
+							deleteKey => {
+								const deleteItemsList = chunk(
+									tables[table][deleteKey].map(item =>
+										copyInputItem(item as INodeExecutionData, [deleteKey]),
+									),
+									1000,
+								);
+								const queryQueue = deleteItemsList.map(deleteValues => {
+									return pool
+										.request()
+										.query(
+											`DELETE FROM ${table} WHERE ${deleteKey} IN ${extractDeleteValues(
+												deleteValues,
+												deleteKey,
+											)};`,
+										);
+								});
+								return Promise.all(queryQueue);
+							},
+						);
+						return Promise.all(deleteKeyResults);
+					}),
+				);
+
+				const rowsAffected = flatten(queriesResults).reduce(
+					(acc: number, resp: mssql.IResult<object>): number =>
+						(acc += resp.rowsAffected.reduce((sum, val) => (sum += val))),
+					0,
+				);
+
+				returnItems = this.helpers.returnJsonArray({
+					rowsAffected,
+				} as IDataObject);
+			} else {
+				await pool.close();
+				throw new Error(`The operation "${operation}" is not supported!`);
+			}
+		} catch (err) {
+			if (this.continueOnFail() === true) {
+				returnItems = items;
+			} else {
+				await pool.close();
+				throw err;
+			}
+		}
+
+		// Close the connection
+		await pool.close();
+
+		return this.prepareOutputData(returnItems);
+	}
+}

packages/nodes-base/nodes/Microsoft/SqlServer/TableInterface.ts

@@ -0,0 +1,7 @@
+import { IDataObject } from 'n8n-workflow';
+
+export interface ITables {
+	[key: string]: {
+		[key: string]: Array<IDataObject>;
+	};
+}

packages/nodes-base/nodes/utils/utilities.ts

@@ -0,0 +1,57 @@
+/**
+ * Creates an array of elements split into groups the length of `size`.
+ * If `array` can't be split evenly, the final chunk will be the remaining
+ * elements.
+ *
+ * @param {Array} array The array to process.
+ * @param {number} [size=1] The length of each chunk
+ * @returns {Array} Returns the new array of chunks.
+ * @example
+ *
+ * chunk(['a', 'b', 'c', 'd'], 2)
+ * // => [['a', 'b'], ['c', 'd']]
+ *
+ * chunk(['a', 'b', 'c', 'd'], 3)
+ * // => [['a', 'b', 'c'], ['d']]
+ */
+export function chunk(array: any[], size: number = 1) {
+	const length = array == null ? 0 : array.length;
+	if (!length || size < 1) {
+		return [];
+	}
+	let index = 0;
+	let resIndex = 0;
+	const result = new Array(Math.ceil(length / size));
+
+	while (index < length) {
+		result[resIndex++] = array.slice(index, (index += size));
+	}
+	return result;
+}
+
+/**
+ * Takes a multidimensional array and converts it to a one-dimensional array.
+ *
+ * @param {Array} nestedArray The array to be flattened.
+ * @returns {Array} Returns the new flattened array.
+ * @example
+ *
+ * flatten([['a', 'b'], ['c', 'd']])
+ * // => ['a', 'b', 'c', 'd']
+ *
+ */
+export function flatten(nestedArray: any[][]) {
+	const result = [];
+
+	(function loop(array: any[]) {
+		for (var i = 0; i < array.length; i++) {
+			if (Array.isArray(array[i])) {
+				loop(array[i]);
+			} else {
+				result.push(array[i]);
+			}
+		}
+	})(nestedArray);
+
+	return result;
+}

packages/nodes-base/package.json

@@ -96,6 +96,7 @@
       "dist/credentials/MicrosoftExcelOAuth2Api.credentials.js",
       "dist/credentials/MicrosoftOAuth2Api.credentials.js",
       "dist/credentials/MicrosoftOneDriveOAuth2Api.credentials.js",
+      "dist/credentials/MicrosoftSqlServer.credentials.js",
       "dist/credentials/MoceanApi.credentials.js",
       "dist/credentials/MondayComApi.credentials.js",
       "dist/credentials/MongoDb.credentials.js",
@@ -241,6 +242,7 @@
       "dist/nodes/MessageBird/MessageBird.node.js",
       "dist/nodes/Microsoft/Excel/MicrosoftExcel.node.js",
       "dist/nodes/Microsoft/OneDrive/MicrosoftOneDrive.node.js",
+      "dist/nodes/Microsoft/SqlServer/MicrosoftSqlServer.node.js",
       "dist/nodes/MoveBinaryData.node.js",
       "dist/nodes/Mocean/Mocean.node.js",
       "dist/nodes/MondayCom/MondayCom.node.js",
@@ -322,6 +324,7 @@
     "@types/lodash.set": "^4.3.6",
     "@types/moment-timezone": "^0.5.12",
     "@types/mongodb": "^3.5.4",
+    "@types/mssql": "^6.0.2",
     "@types/node": "^10.10.1",
     "@types/nodemailer": "^6.4.0",
     "@types/redis": "^2.8.11",
@@ -354,6 +357,7 @@
     "moment": "2.24.0",
     "moment-timezone": "^0.5.28",
     "mongodb": "^3.5.5",
+    "mssql": "^6.2.0",
     "mysql2": "^2.0.1",
     "n8n-core": "~0.37.0",
     "nodemailer": "^6.4.6",