Skip to content

Commit a425321

Browse files
ooemperorooemperor
authored
Merge pull request #9 from ooemperor/development
Development
2 parents 7b27c46 + e7b81af commit a425321

File tree

18 files changed

+457
-92
lines changed

18 files changed

+457
-92
lines changed

‎.env‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,10 @@
1-
CGO_enabled=1
2-
timeout_sec=10
3-
name=go-db-etl
4-
batch_size_read=10000
5-
batch_size_writer=1000
1+
CGO_ENABLED=1
2+
TIMEOUT_SEC=10
3+
NAME=go-db-etl
4+
BATCH_SIZE_READ=10000
5+
BATCH_SIZE_WRITER=1000
6+
# true or false
7+
RUN_SRCINB=true
8+
RUN_INBRDV=true
9+
# 0 = Debug, ... , 3 Status , 4 = Silent
10+
ETL_LOGLEVEL=3

‎build/Dockerfile‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@ WORKDIR /app
2323

2424
# Copy binary from builder into lightweight image
2525
COPY --from=builder /app/bin/go-db-etl /app/bin/go-db-etl
26-
COPY .env .env
2726

2827
# Optionally, copy static files if needed
2928
# COPY --from=builder /app/static ./static

‎build/docker-compose.yml‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
services:
22
runner:
3-
image: ghcr.io/ooemperor/go-db-etl:latest
3+
build:
4+
context: ../
5+
dockerfile: ./build/Dockerfile
46
container_name: go-db-etl
57
volumes:
6-
- "../src.test.json:/app/src.test.json"
8+
- "../src.test.json:/app/src.test.json"
9+
env_file:
10+
- ../.env

‎cmd/main.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ func main() {
2020
// TODO: INSERT pre run output here
2121
runner := runner.Runner{}
2222
runner.Init(srcConfig)
23-
23+
runner.Build()
2424
runner.Run()
2525

2626
}

‎pkg/builder/Global.go‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,9 @@ package builder
22

33
import "fmt"
44

5+
/*
6+
BuildTruncateTableSql return truncate script for a given table
7+
*/
58
func BuildTruncateTableSql(schema string, tableName string) (string, error) {
69
var script string
710

@@ -16,3 +19,45 @@ func BuildTruncateTableSql(schema string, tableName string) (string, error) {
1619

1720
return script, nil
1821
}
22+
23+
/*
24+
ScriptSetTableUnlogged constructs script to make table unlogged
25+
*/
26+
func ScriptSetTableUnlogged(schema string, tableName string) (string, error) {
27+
var script string
28+
29+
if schema == "" {
30+
return "", fmt.Errorf("the schema cannot be blank")
31+
}
32+
if tableName == "" {
33+
return "", fmt.Errorf("the tablename cannot be blank")
34+
}
35+
36+
script += fmt.Sprintf("ALTER TABLE %s.%s SET UNLOGGED;", schema, tableName)
37+
return script, nil
38+
}
39+
40+
/*
41+
ScriptSetTableLogged constructs script to make table unlogged
42+
*/
43+
func ScriptSetTableLogged(schema string, tableName string) (string, error) {
44+
var script string
45+
46+
if schema == "" {
47+
return "", fmt.Errorf("the schema cannot be blank")
48+
}
49+
if tableName == "" {
50+
return "", fmt.Errorf("the tablename cannot be blank")
51+
}
52+
script += fmt.Sprintf("ALTER TABLE %s.%s SET LOGGED;", schema, tableName)
53+
return script, nil
54+
}
55+
56+
/*
57+
ScriptTransactionWrapper Wraps string in transction for postgresql
58+
*/
59+
func ScriptTransactionWrapper(query string) string {
60+
prefix := "DO $$ BEGIN"
61+
suffix := "END $$ ;"
62+
return fmt.Sprintf("%v %v %v", prefix, query, suffix)
63+
}

‎pkg/builder/Global_test.go‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,3 +31,84 @@ func TestBuilderGlobalInsertSchemaEmpty(t *testing.T) {
3131
t.Fatalf("incorrect error: %v", err.Error())
3232
}
3333
}
34+
35+
func TestScriptTransactionWrapper(t *testing.T) {
36+
type args struct {
37+
query string
38+
}
39+
tests := []struct {
40+
name string
41+
args args
42+
want string
43+
}{
44+
{name: "InbPackageBuildTest1", args: args{query: "SELECT 1;"}, want: "DO $$ BEGIN SELECT 1; END $$ ;"},
45+
{name: "InbPackageBuildTest2", args: args{query: "TRUNCATE TABLE public.test;"}, want: "DO $$ BEGIN TRUNCATE TABLE public.test; END $$ ;"},
46+
}
47+
for _, tt := range tests {
48+
t.Run(tt.name, func(t *testing.T) {
49+
if got := ScriptTransactionWrapper(tt.args.query); got != tt.want {
50+
t.Errorf("ScriptTransactionWrapper() = %v, want %v", got, tt.want)
51+
}
52+
})
53+
}
54+
}
55+
56+
func TestScriptSetTableLogged(t *testing.T) {
57+
type args struct {
58+
schema string
59+
tableName string
60+
}
61+
tests := []struct {
62+
name string
63+
args args
64+
want string
65+
wantErr bool
66+
}{
67+
{name: "ScriptSetTableLoggedTest1", args: args{schema: "public", tableName: "test"}, want: "ALTER TABLE public.test SET LOGGED;", wantErr: false},
68+
{name: "ScriptSetTableLoggedTest1", args: args{schema: "", tableName: "test"}, want: "", wantErr: true},
69+
{name: "ScriptSetTableLoggedTest1", args: args{schema: "public", tableName: ""}, want: "", wantErr: true},
70+
{name: "ScriptSetTableLoggedTest1", args: args{schema: "", tableName: ""}, want: "", wantErr: true},
71+
}
72+
for _, tt := range tests {
73+
t.Run(tt.name, func(t *testing.T) {
74+
got, err := ScriptSetTableLogged(tt.args.schema, tt.args.tableName)
75+
if (err != nil) != tt.wantErr {
76+
t.Errorf("ScriptSetTableLogged() error = %v, wantErr %v", err, tt.wantErr)
77+
return
78+
}
79+
if got != tt.want {
80+
t.Errorf("ScriptSetTableLogged() got = %v, want %v", got, tt.want)
81+
}
82+
})
83+
}
84+
}
85+
86+
func TestScriptSetTableUnlogged(t *testing.T) {
87+
type args struct {
88+
schema string
89+
tableName string
90+
}
91+
tests := []struct {
92+
name string
93+
args args
94+
want string
95+
wantErr bool
96+
}{
97+
{name: "ScriptSetTableUnloggedTest1", args: args{schema: "public", tableName: "test"}, want: "ALTER TABLE public.test SET UNLOGGED;", wantErr: false},
98+
{name: "ScriptSetTableUnloggedTest2", args: args{schema: "", tableName: "test"}, want: "", wantErr: true},
99+
{name: "ScriptSetTableUnloggedTest3", args: args{schema: "public", tableName: ""}, want: "", wantErr: true},
100+
{name: "ScriptSetTableUnloggedTest4", args: args{schema: "", tableName: ""}, want: "", wantErr: true},
101+
}
102+
for _, tt := range tests {
103+
t.Run(tt.name, func(t *testing.T) {
104+
got, err := ScriptSetTableUnlogged(tt.args.schema, tt.args.tableName)
105+
if (err != nil) != tt.wantErr {
106+
t.Errorf("ScriptSetTableLogged() error = %v, wantErr %v", err, tt.wantErr)
107+
return
108+
}
109+
if got != tt.want {
110+
t.Errorf("ScriptSetTableLogged() got = %v, want %v", got, tt.want)
111+
}
112+
})
113+
}
114+
}

‎pkg/builder/InbRdv.go‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,12 @@
11
package builder
22

3-
import "fmt"
3+
import (
4+
"database/sql"
5+
"fmt"
6+
7+
"github.com/teambenny/goetl"
8+
"github.com/teambenny/goetl/processors"
9+
)
410

511
/*
612
BuildInbRdvSatCurSelect builds the query for getting the values from inb with hash calculation.
@@ -11,7 +17,7 @@ func BuildInbRdvSatCurSelect(tableName string) (string, error) {
1117
return "", fmt.Errorf("the tablename cannot be blank")
1218
}
1319

14-
script += "SELECT NOW(), NULL, decode(md5(CAST(t.* AS text)), ''hex''), t.* "
20+
script += "SELECT NOW() AS load_dts, NULL AS delete_dts, decode(md5(CAST(t.* AS text)), 'hex') AS frh, t.* "
1521
script += fmt.Sprintf("FROM inb.%s AS t;", tableName)
1622

1723
return script, nil
@@ -70,3 +76,9 @@ func BuildInbRdvSatInsertQuery(tableName string) (string, error) {
7076

7177
return script, nil
7278
}
79+
80+
func BuildInbRdvDummySelect(db *sql.DB) goetl.Processor {
81+
script := "SELECT 1"
82+
processor := processors.NewSQLReader(db, script)
83+
return processor
84+
}

‎pkg/builder/InbRdv_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ func TestBuilderInbRdvSelect(t *testing.T) {
1212
t.Fatalf("Error on InbRdvSatCurSelect: %v", err)
1313
}
1414

15-
if script != `SELECT NOW(), NULL, decode(md5(CAST(t.* AS text)), ''hex''), t.* FROM inb.TestTable AS t;` {
15+
if script != `SELECT NOW() AS load_dts, NULL AS delete_dts, decode(md5(CAST(t.* AS text)), 'hex') AS frh, t.* FROM inb.TestTable AS t;` {
1616
t.Fatalf("Select table statement incorrect: %v", script)
1717
}
1818
}

‎pkg/config/Config.go‎

Lines changed: 18 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -16,31 +16,38 @@ type Configuration struct {
1616
Name string
1717
BatchSizeReader int
1818
BatchSizeWriter int
19+
EtlLogLevel int
20+
RunSrcInb bool
21+
RunInbRdv bool
1922
}
2023

2124
/*
2225
Init initializes the config parameters
2326
*/
24-
func (conf *Configuration) Init(envFile string) (*Configuration, error) {
25-
err := godotenv.Load(envFile)
26-
if err != nil {
27-
return conf, err
28-
}
29-
30-
conf.timeout, _ = strconv.ParseInt(os.Getenv("timeout_sec"), 10, 64)
31-
conf.Name = os.Getenv("name")
32-
var batchReader, _ = strconv.ParseInt(os.Getenv("batch_size_read"), 10, 64)
33-
var batchWriter, _ = strconv.ParseInt(os.Getenv("batch_size_writer"), 10, 64)
27+
func (conf *Configuration) Init() (*Configuration, error) {
28+
_ = godotenv.Load()
29+
_ = godotenv.Load(".env")
30+
_ = godotenv.Load("../.env")
31+
_ = godotenv.Load("../../.env")
32+
_ = godotenv.Load("../../../.env")
33+
conf.timeout, _ = strconv.ParseInt(os.Getenv("TIMEOUT_SEC"), 10, 64)
34+
conf.Name = os.Getenv("NAME")
35+
var batchReader, _ = strconv.ParseInt(os.Getenv("BATCH_SIZE_READ"), 10, 64)
36+
var batchWriter, _ = strconv.ParseInt(os.Getenv("BATCH_SIZE_WRITER"), 10, 64)
3437

3538
conf.BatchSizeReader = int(batchReader)
3639
conf.BatchSizeWriter = int(batchWriter)
3740

41+
conf.RunSrcInb, _ = strconv.ParseBool(os.Getenv("RUN_SRCINB"))
42+
conf.RunInbRdv, _ = strconv.ParseBool(os.Getenv("RUN_INBRDV"))
43+
conf.EtlLogLevel, _ = strconv.Atoi(os.Getenv("ETL_LOGLEVEL"))
44+
3845
return conf, nil
3946
}
4047

4148
/*
4249
Config Export the Config object
4350
*/
44-
var Config, _ = (&(Configuration{})).Init(".env")
51+
var Config, _ = (&(Configuration{})).Init()
4552

4653
var SourceConfiguration, _ = sources.BuildSourceConfig("src.json")

‎pkg/config/Config_test.go‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,29 +3,27 @@ package config
33
import (
44
"reflect"
55
"testing"
6+
7+
"github.com/joho/godotenv"
68
)
79

810
/*
911
TestCases for the config
1012
*/
1113
func TestConfiguration_Init(t *testing.T) {
12-
config := Configuration{timeout: 10, Name: "go-db-etl", BatchSizeReader: 10000, BatchSizeWriter: 1000}
13-
type args struct {
14-
envFile string
15-
}
14+
config := Configuration{timeout: 10, Name: "go-db-etl", BatchSizeReader: 10000, BatchSizeWriter: 1000, EtlLogLevel: 3, RunSrcInb: true, RunInbRdv: true}
1615
tests := []struct {
1716
name string
18-
args args
1917
want *Configuration
2018
wantErr bool
2119
}{
22-
{name: "ConfigTest1", args: args{envFile: "../../.env"}, want: &config, wantErr: false},
23-
{name: "ConfigTest1", args: args{envFile: ".nonExistant"}, want: &Configuration{}, wantErr: true},
20+
{name: "ConfigTest1", want: &config, wantErr: false},
2421
}
2522
for _, tt := range tests {
2623
t.Run(tt.name, func(t *testing.T) {
2724
conf := &Configuration{}
28-
got, err := conf.Init(tt.args.envFile)
25+
_ = godotenv.Load("../../.env")
26+
got, err := conf.Init()
2927
if (err != nil) != tt.wantErr {
3028
t.Errorf("Init() error = %v, wantErr %v", err, tt.wantErr)
3129
return

0 commit comments

Comments
 (0)