refactor sql execution

Author: medcl

config/config.go

@@ -1,7 +1,7 @@
 package config
 
 type AppConfig struct {
-	UIEnabled           bool  `config:"ui_enabled"`
+	UIEnabled bool `config:"ui_enabled"`
 }
 
 var appConfig AppConfig
@@ -13,4 +13,3 @@ func GetAppConfig() AppConfig {
 func SetAppConfig(cfg AppConfig) {
 	appConfig = cfg
 }
-

csv2sql.yml

@@ -11,13 +11,6 @@ plugins:
 - name: csv2sql
   enabled: true
 
-- name: persist
-  enabled: true
-  driver: mysql
-  mysql:
-    connection: root:qeephp@tcp(localhost:3306)/ifish?charset=utf8&parseTime=true&loc=Local
-
-
 modules:
 - name: pipeline
   enabled: true
@@ -53,12 +46,17 @@ modules:
             - feed_pattern
             - produce_pattern
             row_format:
-            - 'INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)'
+            - 'INSERT INTO `ifish`.`aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)'
             - 'VALUES (<{id: }>, <{outer_code: }>, <{common_name: }>, <{scientific_name: }>, <{english_name: }>, <{chinese_name: }>, <{region_name: }>, <{aquatic_category_id: }>);'
 
         process:
         - joint: import_sql
           enabled: true
+          parameters:
+            mysql_conn: root:qeephp@tcp(localhost:3306)/ifish?charset=utf8
+
+        - joint: logging
+          enabled: true
 
         error:
           joint: on_error

executed_sql

@@ -20,9 +20,8 @@ import (
 	_ "expvar"
 	"github.com/infinitbyte/framework"
 	"github.com/infinitbyte/framework/core/module"
-	orm "github.com/infinitbyte/framework/core/persist"
-	q "github.com/infinitbyte/framework/core/queue"
 	pipe "github.com/infinitbyte/framework/core/pipeline"
+	q "github.com/infinitbyte/framework/core/queue"
 	"github.com/infinitbyte/framework/core/util"
 	"github.com/infinitbyte/framework/modules/api"
 	"github.com/infinitbyte/framework/modules/filter"
@@ -34,7 +33,6 @@ import (
 	"github.com/infinitbyte/framework/modules/ui"
 	"github.com/medcl/csv2sql/config"
 	"github.com/medcl/csv2sql/plugin"
-	"github.com/medcl/csv2sql/model"
 )
 
 func main() {
@@ -72,8 +70,6 @@ func main() {
 		//start modules
 		module.Start()
 
-		orm.RegisterSchema(&model.Request{})
-
 		//manual trigger a pipeline event
 		context := pipe.Context{}
 		q.Push("primary", util.ToJSONBytes(context))

main.go

@@ -16,7 +16,6 @@ limitations under the License.
 
 package pipelines
 
-
 import (
 	"github.com/infinitbyte/framework/core/pipeline"
 )
@@ -30,6 +29,5 @@ func (joint ConvertSQLJoint) Name() string {
 
 func (joint ConvertSQLJoint) Process(c *pipeline.Context) error {
 
-
 	return nil
 }

model/request.go

@@ -16,7 +16,6 @@ limitations under the License.
 
 package pipelines
 
-
 import (
 	"github.com/infinitbyte/framework/core/pipeline"
 )
@@ -30,6 +29,5 @@ func (joint DetectMaxIDJoint) Name() string {
 
 func (joint DetectMaxIDJoint) Process(c *pipeline.Context) error {
 
-
 	return nil
 }

pipelines/convert_sql.go

@@ -16,32 +16,66 @@ limitations under the License.
 
 package pipelines
 
-
 import (
-	"github.com/infinitbyte/framework/core/pipeline"
-	"github.com/infinitbyte/framework/core/persist"
+	"database/sql"
 	log "github.com/cihub/seelog"
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/infinitbyte/framework/core/pipeline"
 	"github.com/infinitbyte/framework/core/util"
 )
 
 type ImportSQLJoint struct {
 	pipeline.Parameters
 }
 
+const mysqlConn = "mysql_conn"
+
 func (joint ImportSQLJoint) Name() string {
 	return "import_sql"
 }
 
 func (joint ImportSQLJoint) Process(c *pipeline.Context) error {
 
-	sql:=c.MustGetString(sqlKey)
+	sqlText := c.MustGetString(sqlKey)
+
+	log.Debug("start execute: ", sqlText)
+
+	db, err := sql.Open("mysql", joint.MustGetString(mysqlConn))
+	if err != nil {
+		log.Error(err)
+		panic(err)
+	}
 
-	log.Info("start execute:",sql)
+	tx, err := db.Begin()
+	if err != nil {
+		log.Error(err)
+		panic(err)
+	}
+
+	defer func() {
+		if r := recover(); r != nil {
+			log.Info("the database is rolled back.")
+			err = tx.Rollback()
+			if err != nil {
+				log.Error(err)
+				panic(err)
+			}
+		}
+	}()
+
+	//插入数据
+	result, err := tx.Exec(sqlText)
+
+	if err != nil {
+		log.Error(err, result)
+		panic(err)
+	}
 
-	util.FileAppendNewLine("executed_sql",sql)
+	log.Debug(result)
 
-	err:=persist.RawQuery(sql)
-	if(err!=nil){
+	err = tx.Commit()
+	if err != nil {
+		log.Error(err)
 		panic(err)
 	}
 

pipelines/detect_max_id.go

@@ -2,6 +2,8 @@ package pipelines
 
 import (
 	"github.com/infinitbyte/framework/core/pipeline"
+	"github.com/infinitbyte/framework/core/util"
+	"time"
 )
 
 type LoggingJoint struct {
@@ -13,6 +15,12 @@ func (joint LoggingJoint) Name() string {
 
 func (joint LoggingJoint) Process(c *pipeline.Context) error {
 
+	sql := c.MustGetString(sqlKey)
+
+	util.FileAppendNewLine("log/executed_sql.log", "")
+	util.FileAppendNewLine("log/executed_sql.log", time.Now().String())
+	util.FileAppendNewLine("log/executed_sql.log", sql)
+	util.FileAppendNewLine("log/executed_sql.log", "")
 
 	return nil
 }