feat: initial import

.github/workflows/go.yml

@@ -0,0 +1,25 @@
+name: Go
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+
+  build:
+    name: Build
+    runs-on: ubuntu-latest
+    steps:
+
+    - name: Set up Go 1.x
+      uses: actions/setup-go@v2
+      with:
+        go-version: ^1.15
+
+    - name: Check out code into the Go module directory
+      uses: actions/checkout@v2
+
+    - name: Build
+      run: go build -v ./...

.github/workflows/golangci-lint.yml

@@ -0,0 +1,17 @@
+name: golangci-lint
+on:
+  push:
+    tags:
+      - v*
+    branches:
+      - master
+      - main
+  pull_request:
+jobs:
+  golangci:
+    name: lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: golangci-lint
+        uses: golangci/golangci-lint-action@v2

Makefile

@@ -0,0 +1,5 @@
+lint:
+	golangci-lint run
+
+test:
+	go test ./... -v

README.md

@@ -6,12 +6,6 @@ It uses the underlying introspection API from your RDMS to retrieve automaticall
 
 ## Usage
 
-Tunnel backup database:
-
-```console
-ssh -L 5432:localhost:5433 [email protected]
-```
-
 Export configuration to a environment variable:
 
 ```console
@@ -27,11 +21,11 @@ mkdir -p output
 Extract data from backup database:
 
 ```console
-run main.go -dsn "postgresql://ulule:ulule@localhost:5433/ulule" -path output -action load
+go run cmd/mover/main.go -dsn "postgresql://user:password@localhost:5433/dbname" -path output -action extract -query "SELECT * FROM user WHERE id = 1" -table "user"
 ```
 
 Load data to your local database:
 
 ```console
-go run main.go -dsn "postgresql://ulule:ulule@localhost/ulule" -path output -action load
+go run cmd/mover/main.go -dsn "postgresql://user:password@localhost:5433/dbname" -path output -action load
 ```

cmd/mover/main.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+
+	"go.uber.org/zap"
+
+	"github.com/ulule/mover/config"
+	"github.com/ulule/mover/etl"
+)
+
+var (
+	tableName string
+	query     string
+	path      string
+	dsn       string
+	verbose   bool
+	action    string
+)
+
+func main() {
+	flag.StringVar(&query, "query", "", "query to execute")
+	flag.StringVar(&tableName, "table", "", "table name to export")
+	flag.StringVar(&path, "path", "", "directory output")
+	flag.StringVar(&dsn, "dsn", "", "database dsn")
+	flag.StringVar(&action, "action", "", "action to execute")
+	flag.BoolVar(&verbose, "verbose", false, "verbose logs")
+	flag.Parse()
+
+	var (
+		ctx    = context.Background()
+		logger *zap.Logger
+	)
+	if verbose {
+		logger, _ = zap.NewDevelopment()
+	} else {
+		logger, _ = zap.NewProduction()
+
+	}
+	// nolint:errcheck
+	defer logger.Sync()
+
+	var cfg config.Config
+	if conf := os.Getenv("MOVER_CONF"); conf != "" {
+		if err := config.Load(conf, &cfg); err != nil {
+			logger.Error("unable to config", zap.Error(err), zap.String("config_path", conf))
+		}
+	}
+
+	engine, err := etl.NewEngine(ctx, cfg, dsn, logger)
+	if err != nil {
+		logger.Error("unable to initialize engine", zap.Error(err))
+		return
+	}
+	defer func() {
+		if err := engine.Shutdown(ctx); err != nil {
+			logger.Error("unable to shutdown engine", zap.Error(err))
+		}
+	}()
+
+	switch action {
+	case "extract":
+		if err := engine.Extract(ctx, path, tableName, query); err != nil {
+			logger.Error("unable to extract data",
+				zap.Error(err),
+				zap.String("table_name", tableName),
+				zap.String("query", query))
+		}
+	case "load":
+		if err := engine.Load(ctx, path); err != nil {
+			logger.Error("unable to load data",
+				zap.Error(err),
+				zap.String("path", path))
+		}
+	case "describe":
+		table, err := engine.Describe(ctx, tableName)
+		if err != nil {
+			logger.Error("unable to describe table",
+				zap.Error(err),
+				zap.String("table_name", tableName))
+		}
+
+		fmt.Printf("%v+\n", table)
+	}
+}

config/config.go

@@ -0,0 +1,78 @@
+package config
+
+import (
+	"encoding/json"
+	"log"
+	"os"
+
+	"github.com/ulule/mover/dialect"
+)
+
+type Query struct {
+	TableName string        `json:"table_name"`
+	Query     string        `json:"query"`
+	Table     dialect.Table `json:"-"`
+}
+
+type DownloadHTTP struct {
+	BaseURL string `json:"base_url"`
+}
+
+func (d DownloadHTTP) URL(path string) string {
+	return d.BaseURL + path
+}
+
+type Download struct {
+	Type string       `json:"type"`
+	HTTP DownloadHTTP `json:"http"`
+}
+
+type Column struct {
+	Name     string    `json:"name"`
+	Fake     string    `json:"fake"`
+	Unique   bool      `json:"unique"`
+	Replace  *string   `json:"replace"`
+	Sanitize bool      `json:"sanitize"`
+	Download *Download `json:"download"`
+}
+
+type Schema struct {
+	TableName     string        `json:"table_name"`
+	ReferenceKeys []string      `json:"reference_keys"`
+	Queries       []Query       `json:"queries"`
+	Columns       []Column      `json:"columns"`
+	Table         dialect.Table `json:"-"`
+}
+
+type Config struct {
+	Locale string   `json:"locale"`
+	Schema []Schema `json:"schema"`
+	Extra  []Schema `json:"extra"`
+}
+
+// Load loads the configuration from configuration file path.
+func Load(path string, out interface{}) error {
+	var err error
+
+	f, err := os.Open(path)
+	if f != nil {
+		defer func() {
+			ferr := f.Close()
+			if ferr != nil {
+				log.Println(ferr)
+			}
+		}()
+	}
+
+	if err != nil {
+		return err
+	}
+
+	dec := json.NewDecoder(f)
+	err = dec.Decode(out)
+	if err != nil {
+		return err
+	}
+
+	return err
+}

dialect/dialect.go

@@ -5,8 +5,10 @@ import (
 	"fmt"
 )
 
+// Tables contains a set of tables.
 type Tables []Table
 
+// Get returns a Table with its name.
 func (t Tables) Get(tableName string) Table {
 	for i := range t {
 		if t[i].Name == tableName {
@@ -17,6 +19,7 @@ func (t Tables) Get(tableName string) Table {
 	return Table{}
 }
 
+// Table contains the definition of a database table.
 type Table struct {
 	Name          string
 	PrimaryKeys   []PrimaryKey
@@ -25,8 +28,15 @@ type Table struct {
 	ReferenceKeys ReferenceKeys
 }
 
+// PrimaryKeyColumnName returns the primary key column name.
+func (t Table) PrimaryKeyColumnName() string {
+	return t.PrimaryKeys[0].Name
+}
+
+// Columns contains a set of columns.
 type Columns []Column
 
+// Get returns a Column with its name.
 func (c Columns) Get(name string) Column {
 	for i := range c {
 		if c[i].Name == name {
@@ -41,6 +51,7 @@ func (a Columns) Len() int           { return len(a) }
 func (a Columns) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
 func (a Columns) Less(i, j int) bool { return a[i].Position < a[j].Position }
 
+// Column contains the definition of a column table.
 type Column struct {
 	Name      string
 	Nullable  bool
@@ -49,42 +60,53 @@ type Column struct {
 	Position  int64
 }
 
+// PrimaryKey contains the defintiion of a primary key column.
 type PrimaryKey struct {
 	Name      string
 	DataType  string
 	TableName string
 }
 
+// String returns the string representation of a Primarykey.
 func (f PrimaryKey) String() string {
 	return fmt.Sprintf("%s(%s)", f.TableName, f.Name)
 }
 
+// ForeignKey contains the definition of a foreign key column.
 type ForeignKey struct {
 	Name                 string
 	Definition           string
 	ColumnName           string
 	ReferencedTableName  string
+	ReferencedTable      Table
 	ReferencedColumnName string
 }
 
+// String returns the string representation of a ForeignKey.
 func (f ForeignKey) String() string {
-	return fmt.Sprintf("%s(%s)", f.ReferencedTableName, f.ReferencedColumnName)
+	return fmt.Sprintf("%s(%s)", f.ReferencedTable.Name, f.ReferencedColumnName)
 }
 
+// ForeignKeys contains a set of ForeignKey.
 type ForeignKeys []ForeignKey
 
+// ReferenceKey contains the definition of a reference key.
 type ReferenceKey struct {
 	Name       string
+	Table      Table
 	TableName  string
 	ColumnName string
 }
 
+// String returns the string representation of a ReferenceKey.
 func (f ReferenceKey) String() string {
-	return fmt.Sprintf("%s(%s)", f.TableName, f.ColumnName)
+	return fmt.Sprintf("%s(%s)", f.Table.Name, f.ColumnName)
 }
 
+// ReferenceKeys contains a set of ReferenceKey.
 type ReferenceKeys []ReferenceKey
 
+// Dialect is the main interface to interact with RDMS.
 type Dialect interface {
 	Close(context.Context) error
 	ReferenceKeys(context.Context, string) (ReferenceKeys, error)