|
|
@@ -0,0 +1,648 @@
|
|
|
+/**
|
|
|
+ * @node postgresql-query
|
|
|
+ * @name PostgreSQL Query
|
|
|
+ * @category database
|
|
|
+ * @version 1.0.0
|
|
|
+ * @description Execute CRUD operations on PostgreSQL databases with parameterized queries
|
|
|
+ * @icon database
|
|
|
+ */
|
|
|
+
|
|
|
+const configSchema = {
|
|
|
+ type: 'object',
|
|
|
+ properties: {
|
|
|
+ credentialId: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'PostgreSQL Credential',
|
|
|
+ description: 'Select the PostgreSQL credential to use for connection',
|
|
|
+ dynamicOptions: {
|
|
|
+ source: 'credentials',
|
|
|
+ filter: { type: 'postgresql' }
|
|
|
+ }
|
|
|
+ },
|
|
|
+ database: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Database',
|
|
|
+ description: 'Database name (overrides credential default if set)'
|
|
|
+ },
|
|
|
+ table: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Table',
|
|
|
+ description: 'Table name for query operations'
|
|
|
+ },
|
|
|
+ operation: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Operation',
|
|
|
+ description: 'Type of operation to perform',
|
|
|
+ enum: ['select', 'insert', 'update', 'delete', 'upsert', 'bulkInsert', 'customSql'],
|
|
|
+ enumLabels: ['Select', 'Insert', 'Update', 'Delete', 'Upsert', 'Bulk Insert', 'Custom SQL'],
|
|
|
+ default: 'select'
|
|
|
+ },
|
|
|
+ columns: {
|
|
|
+ type: 'array',
|
|
|
+ title: 'Columns',
|
|
|
+ description: 'Columns to select (empty for all)',
|
|
|
+ items: {
|
|
|
+ type: 'string'
|
|
|
+ },
|
|
|
+ showWhen: { field: 'operation', value: 'select' }
|
|
|
+ },
|
|
|
+ whereConditions: {
|
|
|
+ type: 'array',
|
|
|
+ title: 'WHERE Conditions',
|
|
|
+ description: 'Filter conditions for the query',
|
|
|
+ items: {
|
|
|
+ type: 'object',
|
|
|
+ properties: {
|
|
|
+ column: { type: 'string', title: 'Column' },
|
|
|
+ operator: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Operator',
|
|
|
+ enum: ['=', '!=', '<', '>', '<=', '>=', 'LIKE', 'ILIKE', 'NOT LIKE', 'IN', 'NOT IN', 'IS NULL', 'IS NOT NULL'],
|
|
|
+ default: '='
|
|
|
+ },
|
|
|
+ value: { type: 'string', title: 'Value' }
|
|
|
+ }
|
|
|
+ },
|
|
|
+ showWhen: { field: 'operation', valueIn: ['select', 'update', 'delete'] }
|
|
|
+ },
|
|
|
+ orderBy: {
|
|
|
+ type: 'array',
|
|
|
+ title: 'ORDER BY',
|
|
|
+ description: 'Sorting options',
|
|
|
+ items: {
|
|
|
+ type: 'object',
|
|
|
+ properties: {
|
|
|
+ column: { type: 'string', title: 'Column' },
|
|
|
+ direction: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Direction',
|
|
|
+ enum: ['ASC', 'DESC'],
|
|
|
+ default: 'ASC'
|
|
|
+ }
|
|
|
+ }
|
|
|
+ },
|
|
|
+ showWhen: { field: 'operation', value: 'select' }
|
|
|
+ },
|
|
|
+ limit: {
|
|
|
+ type: 'number',
|
|
|
+ title: 'Limit',
|
|
|
+ description: 'Maximum number of rows to return (0 for no limit)',
|
|
|
+ default: 100,
|
|
|
+ showWhen: { field: 'operation', value: 'select' }
|
|
|
+ },
|
|
|
+ offset: {
|
|
|
+ type: 'number',
|
|
|
+ title: 'Offset',
|
|
|
+ description: 'Number of rows to skip',
|
|
|
+ default: 0,
|
|
|
+ showWhen: { field: 'operation', value: 'select' }
|
|
|
+ },
|
|
|
+ insertData: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Data to Insert (JSON)',
|
|
|
+ description: 'JSON object with column:value pairs. Supports {{variable}} interpolation.',
|
|
|
+ showWhen: { field: 'operation', value: 'insert' }
|
|
|
+ },
|
|
|
+ updateData: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Data to Update (JSON)',
|
|
|
+ description: 'JSON object with column:value pairs to set. Supports {{variable}} interpolation.',
|
|
|
+ showWhen: { field: 'operation', value: 'update' }
|
|
|
+ },
|
|
|
+ upsertData: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Data to Upsert (JSON)',
|
|
|
+ description: 'JSON object with column:value pairs. Supports {{variable}} interpolation.',
|
|
|
+ showWhen: { field: 'operation', value: 'upsert' }
|
|
|
+ },
|
|
|
+ conflictKey: {
|
|
|
+ type: 'array',
|
|
|
+ title: 'Conflict Key Columns',
|
|
|
+ description: 'Columns that form the unique/primary key for conflict detection',
|
|
|
+ items: {
|
|
|
+ type: 'string'
|
|
|
+ },
|
|
|
+ showWhen: { field: 'operation', value: 'upsert' }
|
|
|
+ },
|
|
|
+ updateOnConflict: {
|
|
|
+ type: 'array',
|
|
|
+ title: 'Update Columns on Conflict',
|
|
|
+ description: 'Columns to update when a conflict is detected (empty for all non-key columns)',
|
|
|
+ items: {
|
|
|
+ type: 'string'
|
|
|
+ },
|
|
|
+ showWhen: { field: 'operation', value: 'upsert' }
|
|
|
+ },
|
|
|
+ bulkData: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Bulk Data (JSON Array)',
|
|
|
+ description: 'JSON array of objects to insert. Supports {{variable}} interpolation.',
|
|
|
+ showWhen: { field: 'operation', value: 'bulkInsert' }
|
|
|
+ },
|
|
|
+ batchSize: {
|
|
|
+ type: 'number',
|
|
|
+ title: 'Batch Size',
|
|
|
+ description: 'Number of rows per INSERT statement (0 for all at once)',
|
|
|
+ default: 100,
|
|
|
+ showWhen: { field: 'operation', value: 'bulkInsert' }
|
|
|
+ },
|
|
|
+ customQuery: {
|
|
|
+ type: 'string',
|
|
|
+ title: 'Custom SQL',
|
|
|
+ description: 'Custom SQL query with $1, $2, etc. placeholders for parameters. Supports {{variable}} interpolation.',
|
|
|
+ showWhen: { field: 'operation', value: 'customSql' }
|
|
|
+ },
|
|
|
+ parameters: {
|
|
|
+ type: 'array',
|
|
|
+ title: 'Query Parameters',
|
|
|
+ description: 'Parameters for $1, $2, etc. placeholders in custom SQL',
|
|
|
+ items: {
|
|
|
+ type: 'string'
|
|
|
+ },
|
|
|
+ showWhen: { field: 'operation', value: 'customSql' }
|
|
|
+ }
|
|
|
+ },
|
|
|
+ required: ['credentialId', 'operation']
|
|
|
+};
|
|
|
+
|
|
|
+const inputSchema = {
|
|
|
+ type: 'object',
|
|
|
+ properties: {
|
|
|
+ data: {
|
|
|
+ type: 'any',
|
|
|
+ description: 'Input data available for interpolation in queries'
|
|
|
+ },
|
|
|
+ parameters: {
|
|
|
+ type: 'array',
|
|
|
+ description: 'Override query parameters from input'
|
|
|
+ },
|
|
|
+ rows: {
|
|
|
+ type: 'array',
|
|
|
+ description: 'Array of objects for bulk insert (alternative to bulkData config)'
|
|
|
+ }
|
|
|
+ }
|
|
|
+};
|
|
|
+
|
|
|
+const outputSchema = {
|
|
|
+ type: 'object',
|
|
|
+ properties: {
|
|
|
+ rows: { type: 'array', description: 'Query result rows as array of objects' },
|
|
|
+ metadata: {
|
|
|
+ type: 'object',
|
|
|
+ description: 'Operation metadata',
|
|
|
+ properties: {
|
|
|
+ affectedRows: { type: 'number', description: 'Number of affected rows' },
|
|
|
+ insertId: { type: 'number', description: 'Last auto-generated insert ID (if using RETURNING)' },
|
|
|
+ rowCount: { type: 'number', description: 'Number of rows returned' }
|
|
|
+ }
|
|
|
+ },
|
|
|
+ columns: { type: 'array', description: 'Column names from the result' },
|
|
|
+ query: { type: 'string', description: 'The executed SQL query (for debugging)' }
|
|
|
+ }
|
|
|
+};
|
|
|
+
|
|
|
+const outputs = [
|
|
|
+ { name: 'main', displayName: 'Query Result', type: 'object', color: '#336791' }
|
|
|
+];
|
|
|
+
|
|
|
+function parseJsonData(jsonString, data, fieldName) {
|
|
|
+ if (!jsonString) {
|
|
|
+ throw new Error(fieldName + ' is required');
|
|
|
+ }
|
|
|
+
|
|
|
+ let parsed = jsonString;
|
|
|
+ if (typeof parsed === 'string') {
|
|
|
+ parsed = smartbotic.utils.interpolate(parsed, data);
|
|
|
+ try {
|
|
|
+ parsed = JSON.parse(parsed);
|
|
|
+ } catch (e) {
|
|
|
+ throw new Error(fieldName + ' must be valid JSON: ' + e.message);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return parsed;
|
|
|
+}
|
|
|
+
|
|
|
+function escapeIdentifier(name) {
|
|
|
+ // PostgreSQL uses double quotes for identifiers
|
|
|
+ return '"' + name.replace(/"/g, '""') + '"';
|
|
|
+}
|
|
|
+
|
|
|
+function buildWhereClause(conditions, data, paramOffset) {
|
|
|
+ if (!conditions || conditions.length === 0) {
|
|
|
+ return { clause: '', params: [], nextParam: paramOffset };
|
|
|
+ }
|
|
|
+
|
|
|
+ const clauses = [];
|
|
|
+ const params = [];
|
|
|
+ let paramIdx = paramOffset;
|
|
|
+
|
|
|
+ for (const condition of conditions) {
|
|
|
+ if (!condition.column) continue;
|
|
|
+
|
|
|
+ const col = escapeIdentifier(condition.column);
|
|
|
+ const op = condition.operator || '=';
|
|
|
+ let value = condition.value;
|
|
|
+
|
|
|
+ if (typeof value === 'string') {
|
|
|
+ value = smartbotic.utils.interpolate(value, data);
|
|
|
+ }
|
|
|
+
|
|
|
+ if (op === 'IS NULL') {
|
|
|
+ clauses.push(col + ' IS NULL');
|
|
|
+ } else if (op === 'IS NOT NULL') {
|
|
|
+ clauses.push(col + ' IS NOT NULL');
|
|
|
+ } else if (op === 'IN' || op === 'NOT IN') {
|
|
|
+ let values = value;
|
|
|
+ if (typeof values === 'string') {
|
|
|
+ try {
|
|
|
+ values = JSON.parse(values);
|
|
|
+ } catch (e) {
|
|
|
+ values = values.split(',').map(v => v.trim());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (!Array.isArray(values)) {
|
|
|
+ values = [values];
|
|
|
+ }
|
|
|
+ const placeholders = values.map(() => '$' + (paramIdx++)).join(', ');
|
|
|
+ clauses.push(col + ' ' + op + ' (' + placeholders + ')');
|
|
|
+ params.push(...values);
|
|
|
+ } else {
|
|
|
+ clauses.push(col + ' ' + op + ' $' + (paramIdx++));
|
|
|
+ params.push(value);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return {
|
|
|
+ clause: clauses.length > 0 ? ' WHERE ' + clauses.join(' AND ') : '',
|
|
|
+ params,
|
|
|
+ nextParam: paramIdx
|
|
|
+ };
|
|
|
+}
|
|
|
+
|
|
|
+function buildSelectQuery(config, data) {
|
|
|
+ const table = config.table;
|
|
|
+ if (!table) {
|
|
|
+ throw new Error('Table is required for SELECT operations');
|
|
|
+ }
|
|
|
+
|
|
|
+ const columns = config.columns && config.columns.length > 0
|
|
|
+ ? config.columns.map(escapeIdentifier).join(', ')
|
|
|
+ : '*';
|
|
|
+
|
|
|
+ let query = 'SELECT ' + columns + ' FROM ' + escapeIdentifier(table);
|
|
|
+ const params = [];
|
|
|
+ let paramIdx = 1;
|
|
|
+
|
|
|
+ const where = buildWhereClause(config.whereConditions, data, paramIdx);
|
|
|
+ query += where.clause;
|
|
|
+ params.push(...where.params);
|
|
|
+ paramIdx = where.nextParam;
|
|
|
+
|
|
|
+ if (config.orderBy && config.orderBy.length > 0) {
|
|
|
+ const orderClauses = config.orderBy
|
|
|
+ .filter(o => o.column)
|
|
|
+ .map(o => escapeIdentifier(o.column) + ' ' + (o.direction || 'ASC'));
|
|
|
+ if (orderClauses.length > 0) {
|
|
|
+ query += ' ORDER BY ' + orderClauses.join(', ');
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (config.limit && config.limit > 0) {
|
|
|
+ query += ' LIMIT $' + (paramIdx++);
|
|
|
+ params.push(config.limit);
|
|
|
+
|
|
|
+ if (config.offset && config.offset > 0) {
|
|
|
+ query += ' OFFSET $' + (paramIdx++);
|
|
|
+ params.push(config.offset);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return { query, params };
|
|
|
+}
|
|
|
+
|
|
|
+function buildInsertQuery(config, data) {
|
|
|
+ const table = config.table;
|
|
|
+ if (!table) {
|
|
|
+ throw new Error('Table is required for INSERT operations');
|
|
|
+ }
|
|
|
+
|
|
|
+ const insertData = parseJsonData(config.insertData, data, 'Insert data');
|
|
|
+
|
|
|
+ if (!insertData || typeof insertData !== 'object' || Array.isArray(insertData)) {
|
|
|
+ throw new Error('Insert data must be a JSON object');
|
|
|
+ }
|
|
|
+
|
|
|
+ const columns = Object.keys(insertData);
|
|
|
+ if (columns.length === 0) {
|
|
|
+ throw new Error('Insert data cannot be empty');
|
|
|
+ }
|
|
|
+
|
|
|
+ const values = Object.values(insertData);
|
|
|
+ const placeholders = columns.map((_, i) => '$' + (i + 1)).join(', ');
|
|
|
+
|
|
|
+ // Use RETURNING to get the inserted ID if available
|
|
|
+ const query = 'INSERT INTO ' + escapeIdentifier(table) +
|
|
|
+ ' (' + columns.map(escapeIdentifier).join(', ') + ')' +
|
|
|
+ ' VALUES (' + placeholders + ')' +
|
|
|
+ ' RETURNING *';
|
|
|
+
|
|
|
+ return { query, params: values };
|
|
|
+}
|
|
|
+
|
|
|
+function buildUpdateQuery(config, data) {
|
|
|
+ const table = config.table;
|
|
|
+ if (!table) {
|
|
|
+ throw new Error('Table is required for UPDATE operations');
|
|
|
+ }
|
|
|
+
|
|
|
+ const updateData = parseJsonData(config.updateData, data, 'Update data');
|
|
|
+
|
|
|
+ if (!updateData || typeof updateData !== 'object' || Array.isArray(updateData)) {
|
|
|
+ throw new Error('Update data must be a JSON object');
|
|
|
+ }
|
|
|
+
|
|
|
+ const columns = Object.keys(updateData);
|
|
|
+ if (columns.length === 0) {
|
|
|
+ throw new Error('Update data cannot be empty');
|
|
|
+ }
|
|
|
+
|
|
|
+ const values = Object.values(updateData);
|
|
|
+ let paramIdx = 1;
|
|
|
+ const setClause = columns.map(col => escapeIdentifier(col) + ' = $' + (paramIdx++)).join(', ');
|
|
|
+
|
|
|
+ let query = 'UPDATE ' + escapeIdentifier(table) + ' SET ' + setClause;
|
|
|
+ const params = [...values];
|
|
|
+
|
|
|
+ const where = buildWhereClause(config.whereConditions, data, paramIdx);
|
|
|
+ if (!where.clause) {
|
|
|
+ throw new Error('WHERE conditions are required for UPDATE operations to prevent accidental data modification');
|
|
|
+ }
|
|
|
+ query += where.clause;
|
|
|
+ params.push(...where.params);
|
|
|
+
|
|
|
+ return { query, params };
|
|
|
+}
|
|
|
+
|
|
|
+function buildDeleteQuery(config, data) {
|
|
|
+ const table = config.table;
|
|
|
+ if (!table) {
|
|
|
+ throw new Error('Table is required for DELETE operations');
|
|
|
+ }
|
|
|
+
|
|
|
+ let query = 'DELETE FROM ' + escapeIdentifier(table);
|
|
|
+ const params = [];
|
|
|
+
|
|
|
+ const where = buildWhereClause(config.whereConditions, data, 1);
|
|
|
+ if (!where.clause) {
|
|
|
+ throw new Error('WHERE conditions are required for DELETE operations to prevent accidental data loss');
|
|
|
+ }
|
|
|
+ query += where.clause;
|
|
|
+ params.push(...where.params);
|
|
|
+
|
|
|
+ return { query, params };
|
|
|
+}
|
|
|
+
|
|
|
+function buildUpsertQuery(config, data) {
|
|
|
+ const table = config.table;
|
|
|
+ if (!table) {
|
|
|
+ throw new Error('Table is required for UPSERT operations');
|
|
|
+ }
|
|
|
+
|
|
|
+ const upsertData = parseJsonData(config.upsertData, data, 'Upsert data');
|
|
|
+
|
|
|
+ if (!upsertData || typeof upsertData !== 'object' || Array.isArray(upsertData)) {
|
|
|
+ throw new Error('Upsert data must be a JSON object');
|
|
|
+ }
|
|
|
+
|
|
|
+ const columns = Object.keys(upsertData);
|
|
|
+ if (columns.length === 0) {
|
|
|
+ throw new Error('Upsert data cannot be empty');
|
|
|
+ }
|
|
|
+
|
|
|
+ const values = Object.values(upsertData);
|
|
|
+ const placeholders = columns.map((_, i) => '$' + (i + 1)).join(', ');
|
|
|
+
|
|
|
+ const conflictKeys = config.conflictKey || [];
|
|
|
+ if (conflictKeys.length === 0) {
|
|
|
+ throw new Error('Conflict key columns are required for UPSERT operations in PostgreSQL');
|
|
|
+ }
|
|
|
+
|
|
|
+ let updateColumns = config.updateOnConflict || [];
|
|
|
+
|
|
|
+ if (updateColumns.length === 0) {
|
|
|
+ updateColumns = columns.filter(col => !conflictKeys.includes(col));
|
|
|
+ }
|
|
|
+
|
|
|
+ if (updateColumns.length === 0) {
|
|
|
+ updateColumns = columns;
|
|
|
+ }
|
|
|
+
|
|
|
+ const conflictClause = conflictKeys.map(escapeIdentifier).join(', ');
|
|
|
+ const updateClause = updateColumns
|
|
|
+ .map(col => escapeIdentifier(col) + ' = EXCLUDED.' + escapeIdentifier(col))
|
|
|
+ .join(', ');
|
|
|
+
|
|
|
+ // PostgreSQL ON CONFLICT syntax
|
|
|
+ const query = 'INSERT INTO ' + escapeIdentifier(table) +
|
|
|
+ ' (' + columns.map(escapeIdentifier).join(', ') + ')' +
|
|
|
+ ' VALUES (' + placeholders + ')' +
|
|
|
+ ' ON CONFLICT (' + conflictClause + ')' +
|
|
|
+ ' DO UPDATE SET ' + updateClause +
|
|
|
+ ' RETURNING *';
|
|
|
+
|
|
|
+ return { query, params: values };
|
|
|
+}
|
|
|
+
|
|
|
+function buildBulkInsertQueries(config, data, input) {
|
|
|
+ const table = config.table;
|
|
|
+ if (!table) {
|
|
|
+ throw new Error('Table is required for BULK INSERT operations');
|
|
|
+ }
|
|
|
+
|
|
|
+ let bulkData = input.rows;
|
|
|
+ if (!bulkData) {
|
|
|
+ bulkData = parseJsonData(config.bulkData, data, 'Bulk data');
|
|
|
+ }
|
|
|
+
|
|
|
+ if (!Array.isArray(bulkData)) {
|
|
|
+ throw new Error('Bulk data must be a JSON array of objects');
|
|
|
+ }
|
|
|
+
|
|
|
+ if (bulkData.length === 0) {
|
|
|
+ throw new Error('Bulk data array cannot be empty');
|
|
|
+ }
|
|
|
+
|
|
|
+ const columns = Object.keys(bulkData[0]);
|
|
|
+ if (columns.length === 0) {
|
|
|
+ throw new Error('Bulk data objects cannot be empty');
|
|
|
+ }
|
|
|
+
|
|
|
+ const batchSize = config.batchSize > 0 ? config.batchSize : bulkData.length;
|
|
|
+ const queries = [];
|
|
|
+
|
|
|
+ for (let i = 0; i < bulkData.length; i += batchSize) {
|
|
|
+ const batch = bulkData.slice(i, i + batchSize);
|
|
|
+ const allValues = [];
|
|
|
+ const rowPlaceholders = [];
|
|
|
+ let paramIdx = 1;
|
|
|
+
|
|
|
+ for (const row of batch) {
|
|
|
+ const rowValues = columns.map(col => {
|
|
|
+ const val = row[col];
|
|
|
+ return val !== undefined ? val : null;
|
|
|
+ });
|
|
|
+ allValues.push(...rowValues);
|
|
|
+ const rowParams = columns.map(() => '$' + (paramIdx++));
|
|
|
+ rowPlaceholders.push('(' + rowParams.join(', ') + ')');
|
|
|
+ }
|
|
|
+
|
|
|
+ const query = 'INSERT INTO ' + escapeIdentifier(table) +
|
|
|
+ ' (' + columns.map(escapeIdentifier).join(', ') + ')' +
|
|
|
+ ' VALUES ' + rowPlaceholders.join(', ');
|
|
|
+
|
|
|
+ queries.push({ query, params: allValues, rowCount: batch.length });
|
|
|
+ }
|
|
|
+
|
|
|
+ return queries;
|
|
|
+}
|
|
|
+
|
|
|
+function buildCustomQuery(config, data, input) {
|
|
|
+ if (!config.customQuery) {
|
|
|
+ throw new Error('Custom SQL query is required');
|
|
|
+ }
|
|
|
+
|
|
|
+ const query = smartbotic.utils.interpolate(config.customQuery, data);
|
|
|
+
|
|
|
+ let params = input.parameters || config.parameters || [];
|
|
|
+ if (params.length > 0) {
|
|
|
+ params = params.map(p => {
|
|
|
+ if (typeof p === 'string') {
|
|
|
+ return smartbotic.utils.interpolate(p, data);
|
|
|
+ }
|
|
|
+ return p;
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ return { query, params };
|
|
|
+}
|
|
|
+
|
|
|
+module.exports = {
|
|
|
+ configSchema,
|
|
|
+ inputSchema,
|
|
|
+ outputSchema,
|
|
|
+ outputs,
|
|
|
+
|
|
|
+ async execute(config, input) {
|
|
|
+ const credentialId = config.credentialId;
|
|
|
+ const data = input.data || input;
|
|
|
+
|
|
|
+ if (!credentialId) {
|
|
|
+ throw new Error('PostgreSQL credential is required');
|
|
|
+ }
|
|
|
+
|
|
|
+ const operation = config.operation || 'select';
|
|
|
+ let totalAffectedRows = 0;
|
|
|
+ let lastInsertId = null;
|
|
|
+ let executedQuery = '';
|
|
|
+
|
|
|
+ if (operation === 'bulkInsert') {
|
|
|
+ const queries = buildBulkInsertQueries(config, data, input);
|
|
|
+
|
|
|
+ smartbotic.log.info('PostgreSQL BULK INSERT: ' + queries.length + ' batch(es)');
|
|
|
+
|
|
|
+ for (const queryInfo of queries) {
|
|
|
+ const options = {
|
|
|
+ credentialId,
|
|
|
+ database: config.database,
|
|
|
+ query: queryInfo.query,
|
|
|
+ params: queryInfo.params
|
|
|
+ };
|
|
|
+
|
|
|
+ const result = smartbotic.postgresql.query(options);
|
|
|
+
|
|
|
+ if (!result.success) {
|
|
|
+ smartbotic.log.error('PostgreSQL bulk insert failed: ' + result.error);
|
|
|
+ throw new Error('PostgreSQL bulk insert failed: ' + result.error);
|
|
|
+ }
|
|
|
+
|
|
|
+ totalAffectedRows += result.affectedRows || 0;
|
|
|
+ if (result.insertId) {
|
|
|
+ lastInsertId = result.insertId;
|
|
|
+ }
|
|
|
+ executedQuery = queryInfo.query;
|
|
|
+ }
|
|
|
+
|
|
|
+ smartbotic.log.info('PostgreSQL BULK INSERT completed: ' + totalAffectedRows + ' rows inserted');
|
|
|
+
|
|
|
+ return {
|
|
|
+ rows: [],
|
|
|
+ metadata: {
|
|
|
+ affectedRows: totalAffectedRows,
|
|
|
+ insertId: lastInsertId,
|
|
|
+ rowCount: 0
|
|
|
+ },
|
|
|
+ columns: [],
|
|
|
+ query: executedQuery
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ let queryInfo;
|
|
|
+
|
|
|
+ switch (operation) {
|
|
|
+ case 'select':
|
|
|
+ queryInfo = buildSelectQuery(config, data);
|
|
|
+ break;
|
|
|
+ case 'insert':
|
|
|
+ queryInfo = buildInsertQuery(config, data);
|
|
|
+ break;
|
|
|
+ case 'update':
|
|
|
+ queryInfo = buildUpdateQuery(config, data);
|
|
|
+ break;
|
|
|
+ case 'delete':
|
|
|
+ queryInfo = buildDeleteQuery(config, data);
|
|
|
+ break;
|
|
|
+ case 'upsert':
|
|
|
+ queryInfo = buildUpsertQuery(config, data);
|
|
|
+ break;
|
|
|
+ case 'customSql':
|
|
|
+ queryInfo = buildCustomQuery(config, data, input);
|
|
|
+ break;
|
|
|
+ default:
|
|
|
+ throw new Error('Invalid operation: ' + operation);
|
|
|
+ }
|
|
|
+
|
|
|
+ const { query, params } = queryInfo;
|
|
|
+
|
|
|
+ smartbotic.log.info('PostgreSQL ' + operation.toUpperCase() + ': ' + query.substring(0, 100) + (query.length > 100 ? '...' : ''));
|
|
|
+
|
|
|
+ const options = {
|
|
|
+ credentialId,
|
|
|
+ database: config.database,
|
|
|
+ query,
|
|
|
+ params
|
|
|
+ };
|
|
|
+
|
|
|
+ const result = smartbotic.postgresql.query(options);
|
|
|
+
|
|
|
+ if (!result.success) {
|
|
|
+ smartbotic.log.error('PostgreSQL query failed: ' + result.error);
|
|
|
+ throw new Error('PostgreSQL query failed: ' + result.error);
|
|
|
+ }
|
|
|
+
|
|
|
+ smartbotic.log.info('PostgreSQL ' + operation.toUpperCase() + ' completed: ' +
|
|
|
+ (result.affectedRows || 0) + ' affected, ' +
|
|
|
+ (result.rows || []).length + ' rows returned');
|
|
|
+
|
|
|
+ return {
|
|
|
+ rows: result.rows || [],
|
|
|
+ metadata: {
|
|
|
+ affectedRows: result.affectedRows || 0,
|
|
|
+ insertId: result.insertId || null,
|
|
|
+ rowCount: (result.rows || []).length
|
|
|
+ },
|
|
|
+ columns: result.columns || [],
|
|
|
+ query: query
|
|
|
+ };
|
|
|
+ }
|
|
|
+};
|
