feat: Implement pgx support (#76)

* added pgx v5
* added pgx v4

---------

Co-authored-by: maranqz <[email protected]>

.github/workflows/main.yaml

@@ -24,6 +24,19 @@ jobs:
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_DB: db
+          POSTGRES_USER: user
+          POSTGRES_PASSWORD: pass
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     strategy:
       matrix:
         go-version:
@@ -126,3 +139,5 @@ jobs:
         uses: golangci/golangci-lint-action@v3
         with:
           args: --timeout=3m -v
+          # TODO remove, temporary to restore
+          skip-cache: true

.golangci.yml

@@ -1,6 +1,12 @@
 linters-settings:
   enable-all: true
 
+run:
+  skip-dirs:
+    - trm/mock
+    - trm/manager/mock
+    - sql/mock
+
 linters:
   enable:
     - asasalint

CHANGELOG.md

@@ -5,6 +5,13 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## [1.4.0] - 2023-09-01
+
+### Added
+
+- pgx v4 adapter
+- pgx v5 adapter
+
 ## [1.3.0] - 2023-06-16
 
 ### Added

README.md

@@ -17,6 +17,8 @@ Easiest way to get the perfect repository.
 * [gorm](https://github.com/go-gorm/gorm), [docs](https://pkg.go.dev/github.com/avito-tech/go-transaction-manager/gorm) (Go 1.16)
 * [mongo-go-driver](https://github.com/mongodb/mongo-go-driver), [docs](https://pkg.go.dev/github.com/avito-tech/go-transaction-manager/mongo) (Go 1.13)
 * [go-redis/redis](https://github.com/go-redis/redis), [docs](https://pkg.go.dev/github.com/avito-tech/go-transaction-manager/redis) (Go 1.17)
+* [pgx_v4](https://github.com/jackc/pgx/tree/v4), [docs](https://pkg.go.dev/github.com/avito-tech/go-transaction-manager/pgxv4) (Go 1.16)
+* [pgx_v5](https://github.com/jackc/pgx), [docs](https://pkg.go.dev/github.com/avito-tech/go-transaction-manager/pgxv5) (Go 1.19)
 
 ## Installation
 
@@ -46,6 +48,8 @@ Compatibility beyond that is not guaranteed.
 * [gorm](gorm/example_test.go)
 * [mongo-go-driver](mongo/example_test.go)
 * [go-redis/redis](redis/example_test.go)
+* [pgx_v4](pgxv4/example_test.go)
+* [pgx_v5](pgxv5/example_test.go)
 
 
 Below is an example how to start usage.

go.mod

@@ -7,15 +7,18 @@ require (
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/go-redis/redismock/v8 v8.11.5
 	github.com/golang/mock v1.6.0
+	github.com/jackc/pgconn v1.14.0
+	github.com/jackc/pgx/v4 v4.18.1
+	github.com/jackc/pgx/v5 v5.4.3
 	github.com/jinzhu/copier v0.3.5
 	github.com/jmoiron/sqlx v1.3.5
-	github.com/kr/pretty v0.1.0 // indirect
 	github.com/mattn/go-sqlite3 v1.14.16
 	github.com/onsi/gomega v1.24.1 // indirect
+	github.com/pashagolub/pgxmock v1.8.0
+	github.com/pashagolub/pgxmock/v2 v2.11.0
 	github.com/stretchr/testify v1.8.2
 	go.mongodb.org/mongo-driver v1.12.1
 	go.uber.org/multierr v1.9.0
-	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
 	gorm.io/driver/mysql v1.5.1
 	gorm.io/driver/sqlite v1.5.1
 	gorm.io/gorm v1.25.3

internal/example/habr/finished/go.mod

@@ -7,8 +7,3 @@ require (
 	github.com/jmoiron/sqlx v1.3.5
 	github.com/mattn/go-sqlite3 v1.14.16
 )
-
-require (
-	go.uber.org/atomic v1.7.0 // indirect
-	go.uber.org/multierr v1.9.0 // indirect
-)

pgxv4/context.go

@@ -0,0 +1,52 @@
+//go:build go1.16
+// +build go1.16
+
+//nolint:ireturn,nolintlint // return Tr for external usage.
+//revive:disable:package-comments
+package pgxv4
+
+import (
+	"context"
+
+	"github.com/avito-tech/go-transaction-manager/trm"
+	trmcontext "github.com/avito-tech/go-transaction-manager/trm/context"
+)
+
+// DefaultCtxGetter is the CtxGetter with settings.DefaultCtxKey.
+//
+//nolint:gochecknoglobals
+var DefaultCtxGetter = NewCtxGetter(trmcontext.DefaultManager)
+
+// CtxGetter gets Tr from trm.СtxManager by casting trm.Transaction to Tr.
+type CtxGetter struct {
+	ctxManager trm.СtxManager
+}
+
+//revive:disable:exported
+func NewCtxGetter(c trm.СtxManager) *CtxGetter {
+	return &CtxGetter{ctxManager: c}
+}
+
+func (c *CtxGetter) DefaultTrOrDB(ctx context.Context, db Tr) Tr {
+	if tr := c.ctxManager.Default(ctx); tr != nil {
+		return c.convert(tr)
+	}
+
+	return db
+}
+
+func (c *CtxGetter) TrOrDB(ctx context.Context, key trm.CtxKey, db Tr) Tr {
+	if tr := c.ctxManager.ByKey(ctx, key); tr != nil {
+		return c.convert(tr)
+	}
+
+	return db
+}
+
+func (c *CtxGetter) convert(tr trm.Transaction) Tr {
+	if tx, ok := tr.Transaction().(Tr); ok {
+		return tx
+	}
+
+	return nil
+}

pgxv4/contract.go

@@ -0,0 +1,32 @@
+//go:build go1.16
+// +build go1.16
+
+// Package pgxv4 is an implementation of trm.Transaction interface by Transaction for pgx.Tx.
+package pgxv4
+
+import (
+	"context"
+
+	"github.com/jackc/pgconn"
+	"github.com/jackc/pgx/v4"
+)
+
+// Tr is an interface to work with pgx.Conn, pgxpool.Conn or pgxpool.Pool
+// StmtContext and Stmt are not implemented!
+type Tr interface {
+	Begin(ctx context.Context) (pgx.Tx, error)
+	BeginFunc(ctx context.Context, f func(pgx.Tx) error) (err error)
+
+	CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
+	SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
+
+	Exec(ctx context.Context, sql string, arguments ...interface{}) (commandTag pgconn.CommandTag, err error)
+	Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)
+	QueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row
+	QueryFunc(ctx context.Context, sql string, args []interface{}, scans []interface{}, f func(pgx.QueryFuncRow) error) (pgconn.CommandTag, error)
+}
+
+// Transactional is an interface to work with pgx.Conn, pgxpool.Conn or pgxpool.Pool.
+type Transactional interface {
+	BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
+}

pgxv4/example_test.go

@@ -0,0 +1,124 @@
+//go:build with_real_db
+// +build with_real_db
+
+package pgxv4_test
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v4/pgxpool"
+
+	trmpgx "github.com/avito-tech/go-transaction-manager/pgxv4"
+	"github.com/avito-tech/go-transaction-manager/trm/manager"
+)
+
+// Example demonstrates the implementation of the Repository pattern by trm.Manager.
+func Example() {
+	ctx := context.Background()
+
+	uri := fmt.Sprintf("postgres://%s:%s@%s:%d/%s",
+		"user", "pass", "localhost", 5432, "db",
+	)
+
+	pool, err := pgxpool.Connect(ctx, uri)
+	checkErr(err)
+
+	defer pool.Close()
+
+	sqlStmt := `CREATE TABLE IF NOT EXISTS users_v4 (user_id serial, username TEXT)`
+	_, err = pool.Exec(ctx, sqlStmt)
+	checkErr(err, sqlStmt)
+
+	r := newRepo(pool, trmpgx.DefaultCtxGetter)
+	trManager := manager.Must(trmpgx.NewDefaultFactory(pool))
+
+	u := &user{
+		Username: "username",
+	}
+
+	err = trManager.Do(ctx, func(ctx context.Context) error {
+		if err := r.Save(ctx, u); err != nil {
+			return err
+		}
+
+		return trManager.Do(ctx, func(ctx context.Context) error {
+			u.Username = "new_username"
+
+			return r.Save(ctx, u)
+		})
+	})
+	checkErr(err)
+
+	userFromDB, err := r.GetByID(ctx, u.ID)
+	checkErr(err)
+
+	fmt.Println(userFromDB)
+
+	// Output: &{1 new_username}
+}
+
+type repo struct {
+	db     *pgxpool.Pool
+	getter *trmpgx.CtxGetter
+}
+
+func newRepo(db *pgxpool.Pool, c *trmpgx.CtxGetter) *repo {
+	repo := &repo{
+		db:     db,
+		getter: c,
+	}
+
+	return repo
+}
+
+type user struct {
+	ID       int64
+	Username string
+}
+
+func (r *repo) GetByID(ctx context.Context, id int64) (*user, error) {
+	query := `SELECT * FROM users_v4 WHERE user_id=$1`
+
+	conn := r.getter.DefaultTrOrDB(ctx, r.db)
+	row := conn.QueryRow(ctx, query, id)
+
+	user := &user{}
+
+	err := row.Scan(&user.ID, &user.Username)
+	if err != nil {
+		return nil, err
+	}
+
+	return user, nil
+}
+
+func (r *repo) Save(ctx context.Context, u *user) error {
+	isNew := u.ID == 0
+	conn := r.getter.DefaultTrOrDB(ctx, r.db)
+
+	if !isNew {
+		query := `UPDATE users_v4 SET username = $1 WHERE user_id = $2`
+
+		if _, err := conn.Exec(ctx, query, u.Username, u.ID); err != nil {
+			return err
+		}
+
+		return nil
+	}
+
+	query := `INSERT INTO users_v4 (username) VALUES ($1) RETURNING user_id`
+
+	err := conn.QueryRow(ctx, query, u.Username).Scan(&u.ID)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func checkErr(err error, args ...interface{}) {
+	if err != nil {
+		panic(fmt.Sprint(append([]interface{}{err}, args...)...))
+	}
+}

pgxv4/factory.go

@@ -0,0 +1,24 @@
+//go:build go1.16
+// +build go1.16
+
+package pgxv4
+
+import (
+	"context"
+
+	"github.com/avito-tech/go-transaction-manager/trm"
+)
+
+// NewDefaultFactory creates default trm.Transaction(pgx.Tx).
+func NewDefaultFactory(db Transactional) trm.TrFactory {
+	return NewFactory(db)
+}
+
+// NewFactory creates trm.Transaction(pgx.Tx).
+func NewFactory(db Transactional) trm.TrFactory {
+	return func(ctx context.Context, trms trm.Settings) (context.Context, trm.Transaction, error) {
+		s, _ := trms.(Settings)
+
+		return NewTransaction(ctx, s.TxOpts(), db)
+	}
+}