nettools/pg/pg.go

package pgUtils

import (
	"context"
	"fmt"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/matchsystems/werr"
	"strings"
)

type PgxQuerier interface {
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
}

// Query executes query on a provided querier and tries to parse db response
// Works only with structs
//
// Usage:
//
//	type User struct {
//	  Id   int    `db:"id"`
//	  Name string `db:"name"`
//	}
//
//	db := pgx.Connect(context.Background(), "<url>")
//	users, err := pgUtils.Query[User](context.Background(), db, "SELECT id, name FROM users")
func Query[T any](ctx context.Context, db PgxQuerier, query string, args ...any) (out []T, err error) {
	rows, err := db.Query(ctx, query, args)
	if err != nil {
		return nil, err
	}
	entities, err := pgx.CollectRows(rows, pgx.RowToStructByNameLax[T])
	if err != nil {
		return nil, werr.Wrapf(err, "failed to parse query results")
	}
	return entities, nil
}

// Tx creates new transaction. Cancels it if returned not nil err
func Tx(ctx context.Context, db *pgxpool.Pool, exec func(ctx context.Context, tx pgx.Tx) error) error {
	tx, err := db.Begin(ctx)
	if err != nil {
		return err
	}
	err = exec(ctx, tx)
	if err != nil {
		_ = tx.Rollback(ctx)
		return err
	}
	return tx.Commit(ctx)
}

// SqlInsertMultiple Returns an SQL query to insert multiple items into a table
func SqlInsertMultiple(tableName string, columnNames []string, itemsCount int) string {
	columnsCount := len(columnNames)
	query := strings.Builder{}
	query.WriteString("INSERT INTO ")
	query.WriteString(tableName)
	query.WriteString(" (")
	query.WriteString(strings.Join(columnNames, ","))
	query.WriteString(") VALUES ")

	for item := 0; item < itemsCount; item++ {
		query.WriteByte('(')
		for column := 0; column < columnsCount; column++ {
			query.WriteString(fmt.Sprintf("$%d", item*columnsCount+column))
			if column != columnsCount-1 {
				query.WriteString(", ")
			}
		}
		query.WriteByte(')')
		if item != itemsCount-1 {
			query.WriteByte(',')
		}
	}

	return query.String()
}

// InsertMultiple Inserts multiple items into a specified table
func InsertMultiple(ctx context.Context, db PgxQuerier, tableName string, columnNames []string, items [][]any) (pgconn.CommandTag, error) {
	query := SqlInsertMultiple(tableName, columnNames, len(items))
	itemsJoined := make([]any, 0)
	for _, item := range items {
		itemsJoined = append(itemsJoined, item...)
	}
	return db.Exec(ctx, query, itemsJoined...)
}

// SqlOrder returns "ASC" | "DESC"
// depending on asc param
func SqlOrder(asc bool) string {
	if asc {
		return "ASC"
	}
	return "DESC"
}

func SqlOrderLimit(orderBy string, asc bool, limit int64, offset int64) string {
	query := strings.Builder{}
	query.WriteString("ORDER BY ")
	query.WriteString(orderBy)
	query.WriteByte(' ')
	query.WriteString(SqlOrder(asc))
	query.WriteString(fmt.Sprintf("LIMIT %d OFFSET %d", limit, offset))
	return query.String()
}