feat(push): add import types for push

go.mod

@@ -5,7 +5,7 @@ go 1.17
 require (
 	github.com/Trendyol/overlog v0.1.0
 	github.com/awalterschulze/gographviz v2.0.3+incompatible
-	github.com/cgi-fr/jsonline v0.4.0
+	github.com/cgi-fr/jsonline v0.5.0
 	github.com/docker/docker-credential-helpers v0.6.4
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/godror/godror v0.28.0

go.sum

@@ -59,8 +59,8 @@ github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kB
 github.com/bketelsen/crypt v0.0.4/go.mod h1:aI6NrJ0pMGgvZKL1iVgXLnfIFJtfV+bKCoqOes/6LfM=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
-github.com/cgi-fr/jsonline v0.4.0 h1:Y64IVccv/dmcuY782g95LsbUXptIwk380QrQm3BQNXs=
-github.com/cgi-fr/jsonline v0.4.0/go.mod h1:2eo1zPtPXeGiGCEI+Y2m0GYlQgmRikVeQOwLwOZtWXQ=
+github.com/cgi-fr/jsonline v0.5.0 h1:UFLDJauppXLXWlUXFgAKqlZaKt0g1gRq3tLJUj95ohY=
+github.com/cgi-fr/jsonline v0.5.0/go.mod h1:2eo1zPtPXeGiGCEI+Y2m0GYlQgmRikVeQOwLwOZtWXQ=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=

internal/infra/push/datadestination_sql.go

@@ -314,9 +314,13 @@ func (rw *SQLRowWriter) Write(row push.Row) *push.Error {
 		return err1
 	}
 
+	importedRow := rw.table.Import(row)
 	values := []interface{}{}
 	for _, h := range rw.headers {
-		values = append(values, rw.dd.dialect.ConvertValue(row[h]))
+		value := importedRow.GetOrNil(h)
+
+		v := rw.dd.dialect.ConvertValue(value)
+		values = append(values, v)
 	}
 	log.Trace().Strs("headers", rw.headers).Msg(fmt.Sprint(values))
 

pkg/push/model.go

@@ -29,6 +29,7 @@ type Table interface {
 	Name() string
 	PrimaryKey() []string
 	Columns() ColumnList
+	Import(map[string]interface{}) ImportedRow
 }
 
 // ColumnList is a list of columns.
@@ -62,7 +63,7 @@ type Relation interface {
 type Value interface{}
 
 // Row of data.
-type Row map[string]Value
+type Row map[string]interface{}
 
 // Error is the error type returned by the domain
 type Error struct {

pkg/push/model_table.go

@@ -19,13 +19,19 @@ package push
 
 import (
 	"fmt"
+	"os"
 	"strings"
+
+	"github.com/cgi-fr/jsonline/pkg/jsonline"
+	"github.com/rs/zerolog/log"
 )
 
 type table struct {
 	name    string
 	pk      []string
 	columns ColumnList
+
+	template jsonline.Template
 }
 
 // NewTable initialize a new Table object
@@ -77,3 +83,53 @@ func NewColumn(name string, imp string) Column {
 
 func (c column) Name() string   { return c.name }
 func (c column) Import() string { return c.imp }
+
+type ImportedRow struct {
+	jsonline.Row
+}
+
+func (t *table) initTemplate() {
+	t.template = jsonline.NewTemplate()
+
+	if t.columns == nil {
+		return
+	}
+
+	if l := int(t.columns.Len()); l > 0 {
+		for idx := 0; idx < l; idx++ {
+			col := t.columns.Column(uint(idx))
+			key := col.Name()
+
+			switch col.Import() {
+			case "string":
+				t.template.WithString(key)
+			case "numeric":
+				t.template.WithNumeric(key)
+			case "base64":
+				t.template.WithBinary(key)
+			case "datetime":
+				t.template.WithDateTime(key)
+			case "timestamp":
+				t.template.WithTimestamp(key)
+			case "no":
+				t.template.WithHidden(key)
+			default:
+				t.template.WithAuto(key)
+			}
+		}
+	}
+}
+
+func (t table) Import(row map[string]interface{}) ImportedRow {
+	if t.template == nil {
+		t.initTemplate()
+	}
+
+	result := ImportedRow{t.template.CreateRowEmpty()}
+	err := result.Import(row)
+	if err != nil {
+		log.Err(err).Msg("End pimo")
+		os.Exit(4)
+	}
+	return result
+}