feat: Migrating to bulk update for faster import

go.mod

@@ -2,4 +2,4 @@ module github.com/ViBiOh/spurf/v2
 
 go 1.14
 
-require github.com/ViBiOh/httputils/v3 v3.21.0
+require github.com/ViBiOh/httputils/v3 v3.22.0

go.sum

@@ -1,7 +1,7 @@
 github.com/DATA-DOG/go-sqlmock v1.4.1 h1:ThlnYciV1iM/V0OSF/dtkqWb6xo5qITT1TJBG1MRDJM=
 github.com/DATA-DOG/go-sqlmock v1.4.1/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
-github.com/ViBiOh/httputils/v3 v3.21.0 h1:ST64N5xLQDMCK5ipQ3Sm14pYa/0cmwxbENuz8ZmBT9U=
-github.com/ViBiOh/httputils/v3 v3.21.0/go.mod h1:zGAsgsanf8SlTNWfQ/7taOTOJSYax2b7krZ9RCaQRgc=
+github.com/ViBiOh/httputils/v3 v3.22.0 h1:1gCBclILIMOSVJx+vcEXCVCBm2c0MA1GD/YR8q1+29M=
+github.com/ViBiOh/httputils/v3 v3.22.0/go.mod h1:zGAsgsanf8SlTNWfQ/7taOTOJSYax2b7krZ9RCaQRgc=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=

pkg/enedis/db.go

@@ -3,7 +3,6 @@ package enedis
 import (
 	"context"
 	"database/sql"
-	"fmt"
 	"time"
 
 	"github.com/ViBiOh/httputils/v3/pkg/db"
@@ -18,7 +17,7 @@ WHERE
   name = $1;
 `
 
-func (a *app) getLastFetch(ctx context.Context) (lastTimestamp time.Time, err error) {
+func (a app) getLastFetch(ctx context.Context) (lastTimestamp time.Time, err error) {
 	scanner := func(row *sql.Row) error {
 		return row.Scan(&lastTimestamp)
 	}
@@ -27,26 +26,21 @@ func (a *app) getLastFetch(ctx context.Context) (lastTimestamp time.Time, err er
 	return
 }
 
-const insertQuery = `
-INSERT INTO
-  spurf.enedis_value
-(
-  name,
-  ts,
-  value
-) VALUES (
-  $1,
-  to_timestamp($2),
-  $3
-);
-`
+func (a app) save(ctx context.Context, datas []Value) error {
+	return db.DoAtomic(ctx, a.db, func(ctx context.Context) error {
+		var index int
+		feeder := func(stmt *sql.Stmt) error {
+			if index == len(datas) {
+				return db.ErrBulkEnded
+			}
 
-func (a *app) saveValue(ctx context.Context, o Value) (err error) {
-	err = db.Exec(ctx, insertQuery, a.name, o.Timestamp, o.Valeur)
-	if err != nil {
-		err = fmt.Errorf("unable to save %#v: %w", o, err)
-		return
-	}
+			data := datas[index]
+			index++
 
-	return
+			_, err := stmt.Exec(a.name, data.Timestamp, data.Valeur)
+			return err
+		}
+
+		return db.Bulk(ctx, feeder, "spurf", "enedis_value", "name", "ts", "value")
+	})
 }

pkg/enedis/enedis.go

@@ -12,16 +12,10 @@ import (
 	"strings"
 	"time"
 
-	"github.com/ViBiOh/httputils/v3/pkg/db"
 	"github.com/ViBiOh/httputils/v3/pkg/flags"
 	"github.com/ViBiOh/httputils/v3/pkg/logger"
 )
 
-const (
-	frenchDateFormat = "02/01/2006"
-	isoDateFormat    = "2006-01-02"
-)
-
 // App of package
 type App interface {
 	Start()
@@ -102,18 +96,6 @@ func (a app) handleFile(filename string) error {
 	return nil
 }
 
-func (a app) save(ctx context.Context, datas []Value) error {
-	return db.DoAtomic(ctx, a.db, func(ctx context.Context) error {
-		for _, value := range datas {
-			if err := a.saveValue(ctx, value); err != nil {
-				return err
-			}
-		}
-
-		return nil
-	})
-}
-
 func handleLine(datas []Value, lastInsert time.Time, line string) []Value {
 	parts := strings.Split(line, ";")
 	if len(parts) != 2 {
@@ -140,6 +122,6 @@ func handleLine(datas []Value, lastInsert time.Time, line string) []Value {
 
 	return append(datas, Value{
 		Valeur:    value / 1000,
-		Timestamp: timestamp.Unix(),
+		Timestamp: parts[0],
 	})
 }

pkg/enedis/model.go

@@ -3,6 +3,5 @@ package enedis
 // Value describes data point
 type Value struct {
 	Valeur    float64
-	Ordre     int64
-	Timestamp int64
+	Timestamp string
 }