more db.SQL

db_client/backend/backend.go

@@ -0,0 +1,7 @@
+package backend
+
+import "context"
+
+func GetBackendFromConnectionString(ctx context.Context, connectionString string) (DBClientBackendType, error) {
+	return SqliteDBClientBackend, nil
+}

db_client/backend/dbclientbackendtype.go

@@ -0,0 +1,11 @@
+package backend
+
+//go:generate go run golang.org/x/tools/cmd/stringer -type=DBClientBackendType
+
+type DBClientBackendType int
+
+const (
+	PostgresDBClientBackend DBClientBackendType = iota
+	MySQLDBClientBackend
+	SqliteDBClientBackend
+)

db_client/backend/pgxrowreader.go

@@ -0,0 +1,75 @@
+package backend
+
+import (
+	"fmt"
+	"net/netip"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/turbot/pipe-fittings/queryresult"
+	"github.com/turbot/pipe-fittings/utils"
+)
+
+// PgxRowReader is a RowReader implementation for the pgx database/sql driver
+type PgxRowReader struct{}
+
+func (r *PgxRowReader) Read(columnValues []any, cols []*queryresult.ColumnDef) ([]any, error) {
+	result := make([]any, len(columnValues))
+	for i, columnValue := range columnValues {
+		if columnValue != nil {
+			result[i] = columnValue
+			switch cols[i].DataType {
+			case "_TEXT":
+				if arr, ok := columnValue.([]interface{}); ok {
+					elements := utils.Map(arr, func(e interface{}) string { return e.(string) })
+					result[i] = strings.Join(elements, ",")
+				}
+			case "INET":
+				if inet, ok := columnValue.(netip.Prefix); ok {
+					result[i] = strings.TrimSuffix(inet.String(), "/32")
+				}
+			case "UUID":
+				if bytes, ok := columnValue.([16]uint8); ok {
+					if u, err := uuid.FromBytes(bytes[:]); err == nil {
+						result[i] = u
+					}
+				}
+			case "TIME":
+				if t, ok := columnValue.(pgtype.Time); ok {
+					result[i] = time.UnixMicro(t.Microseconds).UTC().Format("15:04:05")
+				}
+			case "INTERVAL":
+				if interval, ok := columnValue.(pgtype.Interval); ok {
+					var sb strings.Builder
+					years := interval.Months / 12
+					months := interval.Months % 12
+					if years > 0 {
+						sb.WriteString(fmt.Sprintf("%d %s ", years, utils.Pluralize("year", int(years))))
+					}
+					if months > 0 {
+						sb.WriteString(fmt.Sprintf("%d %s ", months, utils.Pluralize("mon", int(months))))
+					}
+					if interval.Days > 0 {
+						sb.WriteString(fmt.Sprintf("%d %s ", interval.Days, utils.Pluralize("day", int(interval.Days))))
+					}
+					if interval.Microseconds > 0 {
+						d := time.Duration(interval.Microseconds) * time.Microsecond
+						formatStr := time.Unix(0, 0).UTC().Add(d).Format("15:04:05")
+						sb.WriteString(formatStr)
+					}
+					result[i] = sb.String()
+				}
+
+			case "NUMERIC":
+				if numeric, ok := columnValue.(pgtype.Numeric); ok {
+					if f, err := numeric.Float64Value(); err == nil {
+						result[i] = f.Float64
+					}
+				}
+			}
+		}
+	}
+	return result, nil
+}

db_client/backend/rowreader.go

@@ -0,0 +1 @@
+package backend

db_client/backend/rowreaderfactory.go

@@ -0,0 +1,22 @@
+package backend
+
+import (
+	"github.com/turbot/pipe-fittings/queryresult"
+)
+
+type RowReader interface {
+	Read(columnValues []any, cols []*queryresult.ColumnDef) ([]any, error)
+}
+
+func RowReaderFactory(backend DBClientBackendType) RowReader {
+	var reader RowReader
+	switch backend {
+	case PostgresDBClientBackend:
+		reader = &PgxRowReader{}
+	case SqliteDBClientBackend:
+		reader = &SqliteRowReader{}
+	default:
+
+	}
+	return reader
+}

db_client/backend/sqliterowreader.go

@@ -0,0 +1,75 @@
+package backend
+
+import (
+	"fmt"
+	"net/netip"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/turbot/pipe-fittings/queryresult"
+	"github.com/turbot/pipe-fittings/utils"
+)
+
+// SqliteRowReader is a RowReader implementation for the sqlite database/sql driver
+type SqliteRowReader struct{}
+
+func (r *SqliteRowReader) Read(columnValues []any, cols []*queryresult.ColumnDef) ([]any, error) {
+	result := make([]any, len(columnValues))
+	for i, columnValue := range columnValues {
+		if columnValue != nil {
+			result[i] = columnValue
+			switch cols[i].DataType {
+			case "_TEXT":
+				if arr, ok := columnValue.([]interface{}); ok {
+					elements := utils.Map(arr, func(e interface{}) string { return e.(string) })
+					result[i] = strings.Join(elements, ",")
+				}
+			case "INET":
+				if inet, ok := columnValue.(netip.Prefix); ok {
+					result[i] = strings.TrimSuffix(inet.String(), "/32")
+				}
+			case "UUID":
+				if bytes, ok := columnValue.([16]uint8); ok {
+					if u, err := uuid.FromBytes(bytes[:]); err == nil {
+						result[i] = u
+					}
+				}
+			case "TIME":
+				if t, ok := columnValue.(pgtype.Time); ok {
+					result[i] = time.UnixMicro(t.Microseconds).UTC().Format("15:04:05")
+				}
+			case "INTERVAL":
+				if interval, ok := columnValue.(pgtype.Interval); ok {
+					var sb strings.Builder
+					years := interval.Months / 12
+					months := interval.Months % 12
+					if years > 0 {
+						sb.WriteString(fmt.Sprintf("%d %s ", years, utils.Pluralize("year", int(years))))
+					}
+					if months > 0 {
+						sb.WriteString(fmt.Sprintf("%d %s ", months, utils.Pluralize("mon", int(months))))
+					}
+					if interval.Days > 0 {
+						sb.WriteString(fmt.Sprintf("%d %s ", interval.Days, utils.Pluralize("day", int(interval.Days))))
+					}
+					if interval.Microseconds > 0 {
+						d := time.Duration(interval.Microseconds) * time.Microsecond
+						formatStr := time.Unix(0, 0).UTC().Add(d).Format("15:04:05")
+						sb.WriteString(formatStr)
+					}
+					result[i] = sb.String()
+				}
+
+			case "NUMERIC":
+				if numeric, ok := columnValue.(pgtype.Numeric); ok {
+					if f, err := numeric.Float64Value(); err == nil {
+						result[i] = f.Float64
+					}
+				}
+			}
+		}
+	}
+	return result, nil
+}

db_client/db_connection_string.go

@@ -0,0 +1,26 @@
+package db_client
+
+import "strings"
+
+func getUseableConnectionString(driver string, connString string) string {
+	if strings.HasPrefix(connString, "sqlite3://") {
+		return strings.TrimPrefix(connString, "sqlite3://")
+	} else if strings.HasPrefix(connString, "sqlite://") {
+		return strings.TrimPrefix(connString, "sqlite://")
+	}
+	return connString
+}
+
+func isPostgresConnectionString(connString string) bool {
+	return strings.HasPrefix(connString, "postgresql://") || strings.HasPrefix(connString, "postgres://")
+}
+
+func isSqliteConnectionString(connString string) bool {
+	return strings.HasPrefix(connString, "sqlite://")
+}
+
+func IsConnectionString(connString string) bool {
+	isPostgres := isPostgresConnectionString(connString)
+	isSqlite := isSqliteConnectionString(connString)
+	return isPostgres || isSqlite
+}

db_client/sql_types.go

@@ -0,0 +1,19 @@
+package db_client
+
+import (
+	"database/sql"
+
+	"github.com/turbot/pipe-fittings/queryresult"
+)
+
+func fieldDescriptionsToColumns(fieldDescriptions []*sql.ColumnType, connection *sql.Conn) []*queryresult.ColumnDef {
+	cols := make([]*queryresult.ColumnDef, len(fieldDescriptions))
+
+	for i, f := range fieldDescriptions {
+		cols[i] = &queryresult.ColumnDef{
+			Name:     f.Name(),
+			DataType: f.DatabaseTypeName(),
+		}
+	}
+	return cols
+}

db_common/column_tag.go

@@ -0,0 +1,27 @@
+package db_common
+
+import (
+	"reflect"
+	"strings"
+)
+
+// ColumnTag is a struct used to display column info in introspection tables
+type ColumnTag struct {
+	Column string
+	// the introspected go type
+	ColumnType string
+}
+
+func newColumnTag(field reflect.StructField) (*ColumnTag, bool) {
+	columnTag, ok := field.Tag.Lookup(TagColumn)
+	if !ok {
+		return nil, false
+	}
+	split := strings.Split(columnTag, ",")
+	if len(split) != 2 {
+		return nil, false
+	}
+	column := split[0]
+	columnType := split[1]
+	return &ColumnTag{column, columnType}, true
+}

db_common/interaces.go

@@ -0,0 +1,10 @@
+package db_common
+
+import (
+	"context"
+	"database/sql"
+)
+
+type ExecContext interface {
+	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
+}