sqlx

Documentation

Overview

Package sqlx is an implementation of trm.Transaction interface by Transaction for sqlx.Tx.

Example

Example demonstrates the implementation of the Repository pattern by trm.Manager.

package main

import (
	"context"
	"fmt"

	"github.com/jmoiron/sqlx"
	_ "github.com/mattn/go-sqlite3"

	trmsqlx "github.com/avito-tech/go-transaction-manager/drivers/sqlx/v2"
	"github.com/avito-tech/go-transaction-manager/trm/v2/manager"
)

// Example demonstrates the implementation of the Repository pattern by trm.Manager.
func main() {
	db := newDB()

	defer db.Close() //nolint:errcheck

	sqlStmt := `CREATE TABLE IF NOT EXISTS user (user_id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, username TEXT);`
	_, err := db.Exec(sqlStmt)
	checkErr(err, sqlStmt)

	r := newRepo(db, trmsqlx.DefaultCtxGetter)

	u := &user{
		Username: "username",
	}

	ctx := context.Background()
	trManager := manager.Must(trmsqlx.NewDefaultFactory(db))

	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)

}

func newDB() *sqlx.DB {
	db, err := sqlx.Open("sqlite3", "file:test?mode=memory")
	checkErr(err)

	return db
}

type repo struct {
	db     *sqlx.DB
	getter *trmsqlx.CtxGetter
}

func newRepo(db *sqlx.DB, c *trmsqlx.CtxGetter) *repo {
	return &repo{
		db:     db,
		getter: c,
	}
}

type user struct {
	ID       int64
	Username string
}

type userRow struct {
	ID       int64  `db:"user_id"`
	Username string `db:"username"`
}

func (r *repo) GetByID(ctx context.Context, id int64) (*user, error) {
	query := "SELECT * FROM user WHERE user_id = ?;"

	row := userRow{}

	err := r.getter.DefaultTrOrDB(ctx, r.db).GetContext(ctx, &row, r.db.Rebind(query), id)
	if err != nil {
		return nil, err
	}

	return r.toModel(row), nil
}

func (r *repo) Save(ctx context.Context, u *user) error {
	isNew := u.ID == 0

	query := `UPDATE user SET username = :username WHERE user_id = :user_id;`
	if isNew {
		query = `INSERT INTO user (username) VALUES (:username);`
	}

	res, err := sqlx.NamedExecContext(
		ctx,
		r.getter.DefaultTrOrDB(ctx, r.db),
		r.db.Rebind(query),
		r.toRow(u),
	)
	if err != nil {
		return err
	} else if !isNew {
		return nil
	} else if u.ID, err = res.LastInsertId(); err != nil {
		return err
	}

	// For PostgreSql need to use NamedQueryContext with RETURNING
	// DO UPDATE SET username = EXCLUDED.username RETURNING id;
	// defer res.Next()
	// if u.ID == 0 && res.Next() {
	//		if err = res.Scan(&u.ID); err != nil {
	//			return err
	//		}
	//	}

	return nil
}

func (r *repo) toRow(model *user) userRow {
	return userRow{
		ID:       model.ID,
		Username: model.Username,
	}
}

func (r *repo) toModel(row userRow) *user {
	return &user{
		ID:       row.ID,
		Username: row.Username,
	}
}

func checkErr(err error, args ...interface{}) {
	if err != nil {
		panic(fmt.Sprint(append([]interface{}{err}, args...)...))
	}
}

Output:

&{1 new_username}

Example (Chained)

Example demonstrates a work of manager.ChainedMW.

// connect DB
db1 := newDB()
defer db1.Close() //nolint:errcheck

db2 := newDB()
defer db2.Close() //nolint:errcheck

// create DB
sqlStmt := `CREATE TABLE IF NOT EXISTS user (user_id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, username TEXT);`
_, err := db1.Exec(sqlStmt)
checkErr(err, sqlStmt)

_, err = db2.Exec(sqlStmt)
checkErr(err, sqlStmt)

// init manager
ctxKey1 := trmcontext.Generate()
m1 := manager.Must(
	trmsqlx.NewDefaultFactory(db1),
	manager.WithSettings(settings.Must(settings.WithCtxKey(ctxKey1))),
)
r1 := newRepo(db1, trmsqlx.NewCtxGetter(trmcontext.New(ctxKey1)))

ctxKey2 := trmcontext.Generate()
m2 := manager.Must(
	trmsqlx.NewDefaultFactory(db2),
	manager.WithSettings(settings.Must(settings.WithCtxKey(ctxKey2))),
)
r2 := newRepo(db2, trmsqlx.NewCtxGetter(trmcontext.New(ctxKey2)))

chainedManager := manager.MustChained([]trm.Manager{m1, m2})

u := &user{Username: "username"}
ctx := context.Background()

err = chainedManager.Do(ctx, func(ctx context.Context) error {
	if err := r1.Save(ctx, u); err != nil {
		return err
	}

	if err := r2.Save(ctx, u); err != nil {
		return err
	}

	return chainedManager.Do(ctx, func(ctx context.Context) error {
		u.Username = "new_username"

		if err = r1.Save(ctx, u); err != nil {
			return err
		}

		return r2.Save(ctx, u)
	})
})
checkErr(err)

userFromDB1, err := r1.GetByID(ctx, u.ID)
checkErr(err)

userFromDB2, err := r1.GetByID(ctx, u.ID)
checkErr(err)

fmt.Println(userFromDB1, userFromDB2)

Output:

&{1 new_username} &{1 new_username}

Index

• Variables
• func NewDefaultFactory(db *sqlx.DB) trm.TrFactory
• func NewFactory(db *sqlx.DB, sp trmsql.SavePoint) trm.TrFactory
• type CtxGetter
  • func NewCtxGetter(c trm.СtxManager) *CtxGetter
  • func (c *CtxGetter) DefaultTrOrDB(ctx context.Context, db Tr) Tr
  • func (c *CtxGetter) TrOrDB(ctx context.Context, key trm.CtxKey, db Tr) Tr
• type Tr
• type Transaction
  • func NewTransaction(ctx context.Context, sp trmsql.SavePoint, opts *sql.TxOptions, db *sqlx.DB) (context.Context, *Transaction, error)
  • func (t *Transaction) Begin(ctx context.Context, _ trm.Settings) (context.Context, trm.Transaction, error)
  • func (t *Transaction) Closed() <-chan struct{}
  • func (t *Transaction) Commit(ctx context.Context) error
  • func (t *Transaction) IsActive() bool
  • func (t *Transaction) Rollback(ctx context.Context) error
  • func (t *Transaction) Transaction() interface{}

Examples

• Package
• Package (Chained)

Variables

var DefaultCtxGetter = NewCtxGetter(trmcontext.DefaultManager)

DefaultCtxGetter is the CtxGetter with settings.DefaultCtxKey.

Functions

func NewDefaultFactory

func NewDefaultFactory(db *sqlx.DB) trm.TrFactory

NewDefaultFactory creates default trm.Transaction(sqlx.Tx).

func NewFactory

func NewFactory(db *sqlx.DB, sp trmsql.SavePoint) trm.TrFactory

NewFactory creates trm.Transaction(sql.Tx).

Types

type CtxGetter

type CtxGetter struct {
	// contains filtered or unexported fields
}

CtxGetter gets Tr from trm.СtxManager by casting trm.Transaction to Tr.

func NewCtxGetter

func NewCtxGetter(c trm.СtxManager) *CtxGetter

NewCtxGetter returns *CtxGetter to get Tr from context.Context.

func (*CtxGetter) DefaultTrOrDB

func (c *CtxGetter) DefaultTrOrDB(ctx context.Context, db Tr) Tr

DefaultTrOrDB returns Tr from context.Context or DB(Tr) otherwise.

func (*CtxGetter) TrOrDB

func (c *CtxGetter) TrOrDB(ctx context.Context, key trm.CtxKey, db Tr) Tr

TrOrDB returns Tr from context.Context by trm.CtxKey or DB(Tr) otherwise.

type Tr

type Tr interface {
	sqlx.ExtContext

	sqlx.Preparer
	Preparex(query string) (*sqlx.Stmt, error)
	PreparexContext(ctx context.Context, query string) (*sqlx.Stmt, error)
	PrepareNamed(query string) (*sqlx.NamedStmt, error)
	PrepareNamedContext(ctx context.Context, query string) (*sqlx.NamedStmt, error)

	sqlx.Execer
	MustExec(query string, args ...interface{}) sql.Result
	MustExecContext(ctx context.Context, query string, args ...interface{}) sql.Result
	NamedExec(query string, arg interface{}) (sql.Result, error)
	NamedExecContext(ctx context.Context, query string, arg interface{}) (sql.Result, error)

	sqlx.Queryer
	QueryRow(query string, args ...interface{}) *sql.Row
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
	NamedQuery(query string, arg interface{}) (*sqlx.Rows, error)

	Select(dest interface{}, query string, args ...interface{}) error
	SelectContext(ctx context.Context, dest interface{}, query string, args ...interface{}) error

	Get(dest interface{}, query string, args ...interface{}) error
	GetContext(ctx context.Context, dest interface{}, query string, args ...interface{}) error
}

Tr is an interface to work with sqlx.DB or sqlx.Tx. Stmtx, StmtxContext, NamedStmt and NamedStmtContext are not implemented!

type Transaction

type Transaction struct {
	// contains filtered or unexported fields
}

Transaction is trm.Transaction for sqlx.Tx.

func NewTransaction

func NewTransaction(
	ctx context.Context,
	sp trmsql.SavePoint,
	opts *sql.TxOptions,
	db *sqlx.DB,
) (context.Context, *Transaction, error)

NewTransaction creates trm.Transaction for sqlx.Tx.

func (*Transaction) Begin

func (t *Transaction) Begin(ctx context.Context, _ trm.Settings) (context.Context, trm.Transaction, error)

Begin nested transaction by save point.

func (*Transaction) Closed

func (t *Transaction) Closed() <-chan struct{}

Closed returns a channel that's closed when transaction committed or rolled back.

func (*Transaction) Commit

func (t *Transaction) Commit(ctx context.Context) error

Commit closes the trm.Transaction.

func (*Transaction) IsActive

func (t *Transaction) IsActive() bool

IsActive returns true if the transaction started but not committed or rolled back.

func (*Transaction) Rollback

func (t *Transaction) Rollback(ctx context.Context) error

Rollback the trm.Transaction.

func (*Transaction) Transaction

func (t *Transaction) Transaction() interface{}

Transaction returns the real transaction sqlx.Tx. trm.NestedTrFactory returns IsActive as true while trm.Transaction is opened.