txmng

README

txmng

txmng is a library for managing transactions at the service layer in a layered architecture.

Installation

$ go get -u github.com/slavaavr/txmng

Quick start

// main.go
var db *pgxpool.Pool = initDB()

dbProvider := txmng.NewPGXProvider(db)
txm, dbm := txmng.New(dbProvider, txmng.WithDefaultRetrier())

r := repo.New(dbm)
s := service.New(txm, r)
s.Do()

// service.go
func (s *Service) Do() error {
    opts := txmng.TxOpts{
        Ctx:       ctx,
        Isolation: txmng.LevelDefault,
        ReadOnly:  false,
        Ext:       nil,
    }
	
    res, err := s.txm.RunTx(opts, func(ctx txmng.Context) (txmng.Result, error) {
        e1, err := s.repo.Do1(ctx, params)
        if err != nil {
            return nil, fmt.Errorf("exec Do1: %w", err)
        }

        e2, err := s.repo.Do2(ctx, params)
        if err != nil {
            return nil, fmt.Errorf("exec Do2: %w", err)
        }
		
        return txmng.NewResult(e1, e2), nil
    })
	if err != nil {
	    return fmt.Errorf("exec tx: %w", err)	
    }
	
    var (
        e1 *Entity1
        e2 *Entity2
    )
	
    if err := res.Scan(&e1, &e2); err != nil {
        return fmt.Errorf("scan: %w", err)
    }
	
    return s.process(e1, e2)
}

// repo.go
func (r *Repo) Do1(ctx txmng.Context, params Params) (*Entity1, error) {
    db, err := r.dbm.GetDB(ctx)
    if err != nil {
        return nil, err	
    }
	
    return r.query(db, params)
}

func (r *Repo) Do2(ctx txmng.Context, params Params) (*Entity2, error) {
    db := r.dbm.MustGetDB(ctx)
    return r.query(db, params)
}

Notes

For more details, see the examples folder.

Documentation

Index

• func New[T any](p DBProvider[T], opts ...Option) (txm TxManager, dbm DBManager[T])
• type Context
• type DBManager
• type DBProvider
  • func NewPGXProvider(db *pgxpool.Pool) DBProvider[PGX]
  • func NewStdSQLProvider(db *sql.DB) DBProvider[StdSQL]
• type IsolationLevel
• type NoTxOpts
• type Option
  • func WithDefaultRetrier() Option
  • func WithRetrier(r Retrier) Option
  • func WithRetrierParams(delays []time.Duration, jitter float64) Option
• type PGX
• type Result
  • func NewResult(args ...any) Result
• type Retrier
• type StdSQL
• type Tx
• type TxManager
• type TxOpts
• type TxOptsExt

Functions

func New

func New[T any](p DBProvider[T], opts ...Option) (txm TxManager, dbm DBManager[T])

Types

type Context

type Context interface {
	context.Context
	// contains filtered or unexported methods
}

type DBManager

type DBManager[T any] interface {
	GetDB(ctx Context) (T, error)
	MustGetDB(ctx Context) T
}

type DBProvider

type DBProvider[T any] interface {
	BeginTx(opts TxOpts) (Tx[T], error)
	GetDB(opts NoTxOpts) T
}

func NewPGXProvider

func NewPGXProvider(db *pgxpool.Pool) DBProvider[PGX]

func NewStdSQLProvider

func NewStdSQLProvider(db *sql.DB) DBProvider[StdSQL]

type IsolationLevel

type IsolationLevel int

const (
	LevelDefault IsolationLevel = iota
	LevelReadUncommitted
	LevelReadCommitted
	LevelWriteCommitted
	LevelRepeatableRead
	LevelSnapshot
	LevelSerializable
	LevelLinearizable
)

type NoTxOpts

type NoTxOpts struct {
	Ctx context.Context
	Ext any
}

type Option

type Option func(cfg *config)

func WithDefaultRetrier

func WithDefaultRetrier() Option

func WithRetrier

func WithRetrier(r Retrier) Option

func WithRetrierParams

func WithRetrierParams(
	delays []time.Duration,
	jitter float64,
) Option

type PGX

type PGX interface {
	pgx.Tx
}

type Result

type Result interface {
	Scan(args ...any) error
}

func NewResult

func NewResult(args ...any) Result

type Retrier

type Retrier interface {
	Do(ctx context.Context, job func() error) error
}

type StdSQL

type StdSQL interface {
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
	PrepareContext(ctx context.Context, query string) (*sql.Stmt, error)
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
	Commit() error
	Rollback() error
}

type Tx

type Tx[T any] interface {
	GetDB() T
	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
}

type TxManager

type TxManager interface {
	RunTx(opts TxOpts, fn func(ctx Context) (Result, error)) (Result, error)
	RunNoTx(opts NoTxOpts, fn func(ctx Context) (Result, error)) (Result, error)
}

type TxOpts

type TxOpts struct {
	Ctx       context.Context
	Isolation IsolationLevel
	ReadOnly  bool
	Ext       any
}

type TxOptsExt

type TxOptsExt struct {
	DeferrableMode bool
	BeginQuery     string
	CommitQuery    string
}