Merge pull request #39 from DEFRA/feature/add-sql-task

Feature/add sql task
DEFRA · Oct 10, 2024 · dc89533 · dc89533
2 parents 9051fa6 + 12fa904
commit dc89533
Show file tree

Hide file tree

Showing 23 changed files with 900 additions and 108 deletions.
diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json
@@ -26,7 +26,8 @@
 				"chris-noring.node-snippets",
 				"cweijan.vscode-postgresql-client2",
 				"mechatroner.rainbow-csv",
-				"esbenp.prettier-vscode"
+				"esbenp.prettier-vscode",
+				"sonarsource.sonarlint-vscode"
 			],
 			"settings": {
 				"jest.runMode": {

diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "ffc-pay-etl-framework",
-  "version": "0.1.14",
+  "version": "1.0.0",
   "publisher": "Defra",
   "main": "dist/cjs/index.js",
   "private": false,

diff --git a/sonar-project.properties b/sonar-project.properties
@@ -1,3 +1,3 @@
-sonar.javascript.exclusions=**/jest.config.js,**/jest.setup.js,**/__mocks__/**,**/node_modules/**,**/test/**,**/test-output/**
+sonar.javascript.exclusions=**/jest.config.js,**/jest.setup.js,**/__mocks__/**,**/node_modules/**,**/test/**,**/test-output/**,/src/examples/**
 sonar.javascript.lcov.reportPaths=test-output/lcov.info
-sonar.exclusions=/test/**,**/*.test.js,*snyk_report.html,*snyk_report.css
+sonar.exclusions=/test/**,**/*.test.js,*snyk_report.html,*snyk_report.css,/src/examples/**
diff --git a/src/database_connections/index.js b/src/database_connections/index.js
@@ -0,0 +1,5 @@
+const { PostgresDatabaseConnection } = require("./postgresDatabaseConnection")
+
+module.exports = {
+  PostgresDatabaseConnection
+}
diff --git a/src/database_connections/postgresDatabaseConnection.js b/src/database_connections/postgresDatabaseConnection.js
@@ -0,0 +1,47 @@
+const DEFAULT_PORT = 5432
+const { Sequelize } = require('sequelize')
+const debug = require('debug')('connection')
+
+/**
+ * 
+ * @param {Object} options 
+ * @param {Object} options.connectionname
+ * @param {Object} options.username
+ * @param {Object} options.password
+ * @param {Object} options.database
+ * @param {Object} options.host
+ * @param {Object} options.port
+ * @returns Connection
+ */
+async function PostgresDatabaseConnection(options){
+    const connectionname = options.connectionname
+    const username = options.username
+    const password = options.password
+    const database = options.database
+    const host = options.host
+    const port = options.port || DEFAULT_PORT
+
+    const sequelize = new Sequelize(database, username, password, {
+      host: host,
+      port: port,
+      dialect: 'postgres',
+      logging: false
+    })
+
+    try{
+      await sequelize.authenticate()
+      debug('sequelize.authenticate succeeded')
+      return {
+        name: connectionname,
+        db: sequelize
+      }
+    } catch(e){
+        debug('sequelize.authenticate failed')
+        debug(e)
+        throw e
+    }
+}
+
+module.exports = {
+  PostgresDatabaseConnection
+}
diff --git a/src/destinations/consoleDestination.js b/src/destinations/consoleDestination.js
@@ -8,14 +8,30 @@ const { Writable } = require("node:stream")
  */
 function ConsoleDestination(options){
     const includeErrors = options.includeErrors
-    return new Writable({
+    const writable = new Writable({
         objectMode: true,
         write(chunk, _, callback){
             if(chunk.errors.length === 0 || includeErrors)
                 console.log(chunk)
+            // @ts-ignore
+            this.tasks?.forEach(task => task.write(chunk))
             callback()
         }
     })
+
+    Object.assign(writable, {
+        setConnection: function (connection){
+            this.connection = connection
+        }.bind(writable),
+        getConnectionName: function (){
+            return this.connection?.name
+        }.bind(writable),
+        setTasks: function(tasks){
+            this.tasks = tasks
+        }.bind(writable)
+    })
+
+    return writable
 }
 
 module.exports = {

diff --git a/src/destinations/csvFileDestination.js b/src/destinations/csvFileDestination.js
@@ -24,7 +24,7 @@ function CSVFileDestination(options){
         fileHandle = fd
     }))
     let headersWritten = false
-    return new Writable({
+    const writable = new Writable({
         objectMode: true,
         write(chunk, _, callback){
             if(!headersWritten && headers) {
@@ -42,6 +42,8 @@ function CSVFileDestination(options){
                     fs.writeFileSync(fileHandle, `${chunk.map(c => `"${c}"`).join(",")}\n`)
                 }
             }
+            // @ts-ignore
+            this.tasks?.forEach(task => task.write(chunk))
             lastChunk = chunk
             callback()
         },
@@ -50,6 +52,20 @@ function CSVFileDestination(options){
             callback()
         }
     })
+
+    Object.assign(writable, {
+        setConnection: function (connection){
+            this.connection = connection
+        }.bind(writable),
+        getConnectionName: function (){
+            return this.connection?.name
+        }.bind(writable),
+        setTasks: function(tasks){
+            this.tasks = tasks
+        }.bind(writable)
+    })
+
+    return writable
 }
 
 util.inherits(CSVFileDestination, EventEmitter)

diff --git a/src/destinations/postgresDestination.js b/src/destinations/postgresDestination.js
@@ -47,49 +47,25 @@ function writeInsertStatement(columnMapping, table, chunk){
  * 
  * @param {Object} options 
  * @param {Object} options.table
- * @param {Object} options.username
- * @param {Object} options.password
- * @param {Object} options.database
- * @param {Object} options.host
- * @param {Object} options.port
+ * @param {Object} options.connectionname
  * @param {Object} options.mapping
  * @param {Object} options.includeErrors
  * @returns Transform
  */
 function PostgresDestination(options){
     EventEmitter.call(this)
     const table = options.table
-    const username = options.username
-    const password = options.password
-    const database = options.database
-    const host = options.host
-    const port = options.port || DEFAULT_PORT
+    const connectionname = options.connectionname
     const mapping = options.mapping
     const includeErrors = options.includeErrors
     let lastChunk
 
-    const sequelize = new Sequelize(database, username, password, {
-        host: host,
-        port: port,
-        dialect: 'postgres',
-        logging: false
-    })
-
-    try{
-        sequelize.authenticate()
-        debug('sequelize.authenticate succeeded')
-    } catch(e){
-        debug('sequelize.authenticate failed')
-        debug(e)
-        throw e
-    }
-
-    return new Transform({
+    const transform = new Transform({
         objectMode: true,
         emitClose: true,
         construct(callback){
-                        // @ts-ignore
-            this.sequelize = sequelize
+            // @ts-ignore
+            this.connectionname = connectionname
             callback()
         },
         write(chunk, _, callback){
@@ -99,12 +75,14 @@ function PostgresDestination(options){
                 insertStatement = writeInsertStatement(mapping, table, chunk)
                 debug('Insert statement: [%s]', insertStatement)
                 // @ts-ignore
-                this.sequelize.query(insertStatement)
+                this.connection.db.query(insertStatement)
                     .then(result => {
                         debug('result %o', result)
                         chunk._result = result
                         lastChunk = chunk
                         // @ts-ignore
+                        this.tasks?.forEach(task => task.write(chunk))
+                        // @ts-ignore
                         callback(null, chunk)
                     }).catch(error => {
                         debug('error %o', error)
@@ -122,6 +100,18 @@ function PostgresDestination(options){
             callback()
         }
     })
+    Object.assign(Transform.prototype, {
+        setConnection: function (connection){
+            this.connection = connection
+        }.bind(transform),
+        getConnectionName: function (){
+            return this.connection?.name
+        }.bind(transform),
+        setTasks: function(tasks){
+            this.tasks = tasks
+        }.bind(transform)
+    })
+    return transform;
 }
 
 module.exports = {

diff --git a/src/destinations/sqlFileDestination.js b/src/destinations/sqlFileDestination.js
@@ -41,9 +41,9 @@ function SQLFileDestination(options){
         })
     }
     function writeUpdateStatement(chunk){
-
+        //TODO
     }
-    return new Writable({
+    const writable = new Writable({
         objectMode: true,
         write(chunk, _, callback){
             if(chunk.errors.length === 0)
@@ -52,15 +52,31 @@ function SQLFileDestination(options){
                     if(chunk.errors.length === 0 | includeErrors){
                         writeInsertStatement(chunk)
                     }
-                } else if (sqlMode === SQL_MODE.UPDATE_MODE){
-                    // @ts-ignore
-                    if(chunk.errors.length === 0 | includeErrors){
-                        writeUpdateStatement(chunk)
-                    }
                 }
+                // TODO - Disabled because Sonar
+                // } else if (sqlMode === SQL_MODE.UPDATE_MODE){
+                //     // @ts-ignore
+                //     if(chunk.errors.length === 0 | includeErrors){
+                //         writeUpdateStatement(chunk)
+                //     }
+                // }
             callback()
         }
     })
+
+    Object.assign(writable, {
+        setConnection: function (connection){
+            this.connection = connection
+        }.bind(writable),
+        getConnectionName: function (){
+            return this.connection?.name
+        }.bind(writable),
+        setTasks: function(tasks){
+            this.tasks = tasks
+        }.bind(writable)
+    })
+
+    return writable
 }
 
 module.exports = {