WIP Sql bindings extensions (postgreSQL)

bindings/postgres/postgres.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"reflect"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -34,14 +35,43 @@ import (
 
 // List of operations.
 const (
-	execOperation  bindings.OperationKind = "exec"
-	queryOperation bindings.OperationKind = "query"
-	closeOperation bindings.OperationKind = "close"
-
+	execOperation       bindings.OperationKind = "exec"
+	queryOperation      bindings.OperationKind = "query"
+	closeOperation      bindings.OperationKind = "close"
+	registerOperation   bindings.OperationKind = "register"
+	saveOperation       bindings.OperationKind = "save"
+	findByIdOperation   bindings.OperationKind = "findById"
+	findAllOperation    bindings.OperationKind = "findByAll"
+	deleteByIdOperation bindings.OperationKind = "deleteById"
+	deleteAllOperation  bindings.OperationKind = "deleteAll"
+	existsByIdOperation bindings.OperationKind = "existsById"
+	countOperation      bindings.OperationKind = "count"
+
+	sqlInsert     = "INSERT INTO %v (%v) VALUES (%v)"
+	sqlUpdateById = "UPDATE '%v' SET %v WHERE %v = %d"
+	sqlSelectById = "SELECT * FROM %v WHERE %v = %d"
+	sqlSelectAll  = "SELECT (%v) FROM %v"
+	sqlDeleteAll  = "DELETE FROM %v"
+	sqlCountAll   = "SELECT COUNT(id) FROM %v"
+
+	operationType = "type"
+	//SQL operations
 	commandSQLKey  = "sql"
 	commandArgsKey = "params"
+
+	//Entity operations
+	commandEntityName  = "entity-name"
+	commandEntityId    = "entity-id"
+	commandEntityProps = "props"
 )
 
+var entities = map[string]Entity{}
+
+type Entity struct {
+	id         string
+	properties []string
+}
+
 // Postgres represents PostgreSQL output binding.
 type Postgres struct {
 	logger logger.Logger
@@ -120,6 +150,10 @@ func (p *Postgres) Operations() []bindings.OperationKind {
 	return []bindings.OperationKind{
 		execOperation,
 		queryOperation,
+		registerOperation,
+		saveOperation,
+		findByIdOperation,
+		deleteByIdOperation,
 		closeOperation,
 	}
 }
@@ -144,19 +178,39 @@ func (p *Postgres) Invoke(ctx context.Context, req *bindings.InvokeRequest) (res
 		return nil, errors.New("metadata required")
 	}
 
-	// Metadata property "sql" contains the query to execute
-	sql := req.Metadata[commandSQLKey]
-	if sql == "" {
-		return nil, fmt.Errorf("required metadata not set: %s", commandSQLKey)
+	operationType := req.Metadata[operationType]
+	if operationType == "" {
+		operationType = "sql"
 	}
 
-	// Metadata property "params" contains JSON-encoded parameters, and it's optional
-	// If present, it must be unserializable into a []any object
+	var sql = ""
 	var args []any
-	if argsStr := req.Metadata[commandArgsKey]; argsStr != "" {
-		err = json.Unmarshal([]byte(argsStr), &args)
-		if err != nil {
-			return nil, fmt.Errorf("invalid metadata property %s: failed to unserialize into an array: %w", commandArgsKey, err)
+	var entityName string
+	var entityProps []string
+	var entityId = ""
+	if operationType == "sql" {
+		// Metadata property "sql" contains the query to execute
+		sql = req.Metadata[commandSQLKey]
+		if sql == "" {
+			return nil, fmt.Errorf("required metadata not set: %s", commandSQLKey)
+		}
+
+		// Metadata property "params" contains JSON-encoded parameters, and it's optional
+		// If present, it must be unserializable into a []any object
+		if argsStr := req.Metadata[commandArgsKey]; argsStr != "" {
+			err = json.Unmarshal([]byte(argsStr), &args)
+			if err != nil {
+				return nil, fmt.Errorf("invalid metadata property %s: failed to unserialize into an array: %w", commandArgsKey, err)
+			}
+		}
+	} else if operationType == "entity" {
+		entityName = req.Metadata[commandEntityName]
+		entityId = req.Metadata[commandEntityId]
+		if argsStr := req.Metadata[commandEntityProps]; argsStr != "" {
+			err = json.Unmarshal([]byte(argsStr), &entityProps)
+			if err != nil {
+				return nil, fmt.Errorf("invalid metadata property %s: failed to unserialize into an array: %w", commandArgsKey, err)
+			}
 		}
 	}
 
@@ -177,6 +231,39 @@ func (p *Postgres) Invoke(ctx context.Context, req *bindings.InvokeRequest) (res
 		}
 		resp.Metadata["rows-affected"] = strconv.FormatInt(r, 10) // 0 if error
 
+	case registerOperation:
+
+		entities[entityName] = Entity{properties: entityProps, id: entityId}
+
+	case findAllOperation:
+		columns := strings.Join(entities[entityName].properties, ",")
+		sql := fmt.Sprintf(sqlSelectAll, columns, entityName)
+		d, err := p.query(ctx, sql, args...)
+		if err != nil {
+			return nil, err
+		}
+		resp.Data = d
+
+	case deleteAllOperation:
+		sql := fmt.Sprintf(sqlDeleteAll, entityName)
+		r, err := p.exec(ctx, sql, args...)
+		if err != nil {
+			return nil, err
+		}
+		resp.Metadata["rows-affected"] = strconv.FormatInt(r, 10) // 0 if error
+
+	case saveOperation:
+		columns := strings.Join(entities[entityName].properties, ",")
+		columnsWithoutId := strings.Replace(columns, fmt.Sprintf("%v,", entities[entityName].id), "", -1)
+		values := strings.Join(entityProps, ",")
+		sql := fmt.Sprintf(sqlInsert, entityName, columnsWithoutId, values)
+
+		r, err := p.exec(ctx, sql, args...) //args are values
+		if err != nil {
+			return nil, err
+		}
+		resp.Metadata["rows-affected"] = strconv.FormatInt(r, 10) // 0 if error
+
 	case queryOperation:
 		d, err := p.query(ctx, sql, args...)
 		if err != nil {

bindings/postgres/postgres_test.go

@@ -40,13 +40,25 @@ const (
 	testSelect = "SELECT * FROM foo WHERE id < 3"
 )
 
+const (
+	testCustomersTableDDL = `CREATE TABLE IF NOT EXISTS customers (
+								customer_id SERIAL NOT NULL PRIMARY KEY,
+								customer_name VARCHAR(255),
+								contact_name VARCHAR(255),
+								address VARCHAR(255),
+								city VARCHAR(255),
+								postal_code VARCHAR(255),
+								country VARCHAR(255)
+								)`
+)
+
 func TestOperations(t *testing.T) {
 	t.Parallel()
 	t.Run("Get operation list", func(t *testing.T) {
 		b := NewPostgres(nil)
 		assert.NotNil(t, b)
 		l := b.Operations()
-		assert.Len(t, l, 3)
+		assert.Len(t, l, 7)
 	})
 }
 
@@ -176,6 +188,102 @@ func testInitConfiguration(t *testing.T, connectionString string) {
 	}
 }
 
+// `go test -v -count=1 ./bindings/postgres -run ^TestPostgresEntityIntegration`
+func TestPostgresEntityIntegration(t *testing.T) {
+	url := os.Getenv("POSTGRES_TEST_CONN_URL")
+	if url == "" {
+		t.SkipNow()
+	}
+
+	t.Run("Test init configurations", func(t *testing.T) {
+		testInitConfiguration(t, url)
+	})
+
+	// live DB test
+	b := NewPostgres(logger.NewLogger("test")).(*Postgres)
+	m := bindings.Metadata{Base: metadata.Base{Properties: map[string]string{"connectionString": url}}}
+	if err := b.Init(context.Background(), m); err != nil {
+		t.Fatal(err)
+	}
+
+	// create table
+	req := &bindings.InvokeRequest{
+		Operation: execOperation,
+		Metadata:  map[string]string{commandSQLKey: testCustomersTableDDL},
+	}
+	ctx := context.TODO()
+	t.Run("Invoke create table", func(t *testing.T) {
+		res, err := b.Invoke(ctx, req)
+		assertResponse(t, res, err)
+	})
+
+	req = &bindings.InvokeRequest{
+		Operation: registerOperation,
+		Metadata: map[string]string{operationType: "entity",
+			commandEntityName: "customers"},
+	}
+	t.Run("Invoke register", func(t *testing.T) {
+		req.Metadata[commandEntityId] = "customer_id"
+		req.Metadata[commandEntityProps] = "[\"customer_id\",\"customer_name\",\"contact_name\",\"address\", \"city\", \"postal_code\",\"country\"]"
+		res, err := b.Invoke(ctx, req)
+		assertResponse(t, res, err)
+	})
+
+	req = &bindings.InvokeRequest{
+		Operation: findAllOperation,
+		Metadata: map[string]string{operationType: "entity",
+			commandEntityName: "customers"},
+	}
+
+	t.Run("Invoke findAll", func(t *testing.T) {
+		res, err := b.Invoke(ctx, req)
+		assertResponse(t, res, err)
+		assert.Equal(t, string(res.Data), "[]")
+	})
+
+	req = &bindings.InvokeRequest{
+		Operation: saveOperation,
+		Metadata: map[string]string{operationType: "entity",
+			commandEntityName: "customers"},
+	}
+	t.Run("Invoke save", func(t *testing.T) {
+		// @TODO: we should marshal a customer struct to JSON, but it will not be an array
+		req.Metadata[commandEntityProps] = "[\"'salaboy'\",\"'salaboy'\",\"'chiswick'\", \"'london'\", \"'w4'\",\"'uk'\"]"
+		res, err := b.Invoke(ctx, req)
+		assertResponse(t, res, err)
+	})
+
+	req = &bindings.InvokeRequest{
+		Operation: findAllOperation,
+		Metadata: map[string]string{operationType: "entity",
+			commandEntityName: "customers"},
+	}
+	t.Run("Invoke findAll", func(t *testing.T) {
+		res, err := b.Invoke(ctx, req)
+		assertResponse(t, res, err)
+		assert.NotNil(t, res.Data)
+		//@TODO: we can marshal this into a customer struct and validate props
+		assert.Contains(t, string(res.Data), "\"salaboy\",\"salaboy\",\"chiswick\",\"london\",\"w4\",\"uk\"")
+	})
+
+	req = &bindings.InvokeRequest{
+		Operation: deleteAllOperation,
+		Metadata: map[string]string{operationType: "entity",
+			commandEntityName: "customers"},
+	}
+	t.Run("Invoke delete all", func(t *testing.T) {
+
+		res, err := b.Invoke(ctx, req)
+		assertResponse(t, res, err)
+		assert.Equal(t, res.Metadata["rows-affected"], "1")
+	})
+
+	t.Run("Close", func(t *testing.T) {
+		err := b.Close()
+		require.NoError(t, err, "expected no error closing output binding")
+	})
+}
+
 func assertResponse(t *testing.T, res *bindings.InvokeResponse, err error) {
 	require.NoError(t, err)
 	assert.NotNil(t, res)