db

Documentation

Index

• Constants
• Variables
• func ConnectDBPool(ctx context.Context, p params.Database) (*pgxpool.Pool, error)
• func GetLastMigrationAvailable() (uint, error)
• func IsSerializationError(err error) bool
• func MigrateDown(params params.Database) error
• func MigrateTo(ctx context.Context, p params.Database, version uint, force bool) error
• func MigrateUp(p params.Database) error
• func MigrateVersion(ctx context.Context, params params.Database) (uint, bool, error)
• func Ping(ctx context.Context, pool *pgxpool.Pool) error
• func Prefix(prefix string) string
• func Select(d Tx, results interface{}, query string, args ...interface{}) error
• func ValidateSchemaUpToDate(ctx context.Context, params params.Database) error
• type Database
  • func BuildDatabaseConnection(ctx context.Context, dbParams params.Database) Database
  • func ConnectDB(ctx context.Context, p params.Database) (Database, error)
• type DatabaseMigrator
  • func NewDatabaseMigrator(params params.Database) *DatabaseMigrator
  • func (d *DatabaseMigrator) Migrate(ctx context.Context) error
• type LoggedRows
  • func Logged(r pgx.Rows, start time.Time, l logging.Logger) *LoggedRows
  • func (lr *LoggedRows) Close()
  • func (lr *LoggedRows) CommandTag() pgconn.CommandTag
  • func (lr *LoggedRows) Err() error
  • func (lr *LoggedRows) FieldDescriptions() []pgproto3.FieldDescription
  • func (lr *LoggedRows) Next() bool
  • func (lr *LoggedRows) RawValues() [][]byte
  • func (lr *LoggedRows) Scan(dest ...interface{}) error
  • func (lr *LoggedRows) Values() ([]interface{}, error)
• type Migrator
• type PgxDatabase
  • func NewPgxDatabase(db *pgxpool.Pool) *PgxDatabase
  • func (d *PgxDatabase) Close()
  • func (d *PgxDatabase) Exec(ctx context.Context, query string, args ...interface{}) (pgconn.CommandTag, error)
  • func (d *PgxDatabase) Get(ctx context.Context, dest interface{}, query string, args ...interface{}) error
  • func (d *PgxDatabase) GetPrimitive(ctx context.Context, dest interface{}, query string, args ...interface{}) error
  • func (d *PgxDatabase) Metadata(ctx context.Context) (map[string]string, error)
  • func (d *PgxDatabase) Pool() *pgxpool.Pool
  • func (d *PgxDatabase) Query(ctx context.Context, query string, args ...interface{}) (rows pgx.Rows, err error)
  • func (d *PgxDatabase) Select(ctx context.Context, results interface{}, query string, args ...interface{}) error
  • func (d *PgxDatabase) Stats() sql.DBStats
  • func (d *PgxDatabase) Transact(ctx context.Context, fn TxFunc, opts ...TxOpt) (interface{}, error)
• type Querier
• type QueryOptions
• type Tx
• type TxFunc
  • func Void(fn func(tx Tx) error) TxFunc
• type TxOpt
  • func ReadCommitted() TxOpt
  • func ReadOnly() TxOpt
  • func RepeatableRead() TxOpt
  • func WithIsolationLevel(level pgx.TxIsoLevel) TxOpt
  • func WithLogger(logger logging.Logger) TxOpt
• type TxOptions
  • func DefaultTxOptions() *TxOptions

Constants

const (
	DefaultMaxOpenConnections    = 25
	DefaultMaxIdleConnections    = 25
	DefaultConnectionMaxLifetime = 5 * time.Minute
	DatabaseDriver               = "pgx"
)

const (
	SerializationRetryMaxAttempts   = 10
	SerializationRetryStartInterval = time.Millisecond * 2
)

Variables

var (
	ErrNotFound      = fmt.Errorf("not found: %w", pgx.ErrNoRows)
	ErrAlreadyExists = errors.New("already exists")
	ErrSerialization = errors.New("serialization error")
)

var ErrSchemaNotCompatible = errors.New("db schema version not compatible with latest version")

Functions

func ConnectDBPool

func ConnectDBPool(ctx context.Context, p params.Database) (*pgxpool.Pool, error)

ConnectDBPool connects to a database using the database params and returns a connection pool

func GetLastMigrationAvailable

func GetLastMigrationAvailable() (uint, error)

func IsSerializationError

func IsSerializationError(err error) bool

func MigrateDown

func MigrateDown(params params.Database) error

func MigrateTo

func MigrateTo(ctx context.Context, p params.Database, version uint, force bool) error

func MigrateUp

func MigrateUp(p params.Database) error

func MigrateVersion

func MigrateVersion(ctx context.Context, params params.Database) (uint, bool, error)

func Ping

func Ping(ctx context.Context, pool *pgxpool.Pool) error

func Prefix

func Prefix(prefix string) string

func Select

func Select(d Tx, results interface{}, query string, args ...interface{}) error

func ValidateSchemaUpToDate

func ValidateSchemaUpToDate(ctx context.Context, params params.Database) error

Types

type Database

type Database interface {
	Querier
	Select(ctx context.Context, dest interface{}, query string, args ...interface{}) error
	Get(ctx context.Context, dest interface{}, query string, args ...interface{}) error
	GetPrimitive(ctx context.Context, dest interface{}, query string, args ...interface{}) error
	Exec(ctx context.Context, query string, args ...interface{}) (pgconn.CommandTag, error)
	Transact(ctx context.Context, fn TxFunc, opts ...TxOpt) (interface{}, error)

	Close()
	Metadata(ctx context.Context) (map[string]string, error)
	Stats() sql.DBStats
	Pool() *pgxpool.Pool
}

func BuildDatabaseConnection

func BuildDatabaseConnection(ctx context.Context, dbParams params.Database) Database

BuildDatabaseConnection returns a database connection based on a pool for the configuration in c.

func ConnectDB

func ConnectDB(ctx context.Context, p params.Database) (Database, error)

ConnectDB connects to a database using the database params and returns Database

type DatabaseMigrator

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

func NewDatabaseMigrator

func NewDatabaseMigrator(params params.Database) *DatabaseMigrator

func (*DatabaseMigrator) Migrate

func (d *DatabaseMigrator) Migrate(ctx context.Context) error

type LoggedRows

type LoggedRows struct {
	pgx.Rows
	Start  time.Time
	Closed bool
	// contains filtered or unexported fields
}

LoggedRows is a pgx.Rows that wraps and traces another pgx.Rows.

func Logged

func Logged(r pgx.Rows, start time.Time, l logging.Logger) *LoggedRows

Logged returns a pgx.Rows that will forward calls to r and logs their durations.

func (*LoggedRows) Close

func (lr *LoggedRows) Close()

func (*LoggedRows) CommandTag

func (lr *LoggedRows) CommandTag() pgconn.CommandTag

func (*LoggedRows) Err

func (lr *LoggedRows) Err() error

func (*LoggedRows) FieldDescriptions

func (lr *LoggedRows) FieldDescriptions() []pgproto3.FieldDescription

func (*LoggedRows) Next

func (lr *LoggedRows) Next() bool

func (*LoggedRows) RawValues

func (lr *LoggedRows) RawValues() [][]byte

func (*LoggedRows) Scan

func (lr *LoggedRows) Scan(dest ...interface{}) error

func (*LoggedRows) Values

func (lr *LoggedRows) Values() ([]interface{}, error)

type Migrator

type Migrator interface {
	Migrate(ctx context.Context) error
}

type PgxDatabase

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

func NewPgxDatabase

func NewPgxDatabase(db *pgxpool.Pool) *PgxDatabase

func (*PgxDatabase) Close

func (d *PgxDatabase) Close()

func (*PgxDatabase) Exec

func (d *PgxDatabase) Exec(ctx context.Context, query string, args ...interface{}) (pgconn.CommandTag, error)

func (*PgxDatabase) Get

func (d *PgxDatabase) Get(ctx context.Context, dest interface{}, query string, args ...interface{}) error

func (*PgxDatabase) GetPrimitive

func (d *PgxDatabase) GetPrimitive(ctx context.Context, dest interface{}, query string, args ...interface{}) error

func (*PgxDatabase) Metadata

func (d *PgxDatabase) Metadata(ctx context.Context) (map[string]string, error)

func (*PgxDatabase) Pool

func (d *PgxDatabase) Pool() *pgxpool.Pool

func (*PgxDatabase) Query

func (d *PgxDatabase) Query(ctx context.Context, query string, args ...interface{}) (rows pgx.Rows, err error)

func (*PgxDatabase) Select

func (d *PgxDatabase) Select(ctx context.Context, results interface{}, query string, args ...interface{}) error

func (*PgxDatabase) Stats

func (d *PgxDatabase) Stats() sql.DBStats

func (*PgxDatabase) Transact

func (d *PgxDatabase) Transact(ctx context.Context, fn TxFunc, opts ...TxOpt) (interface{}, error)

type Querier

type Querier interface {
	Query(ctx context.Context, query string, args ...interface{}) (pgx.Rows, error)
}

type QueryOptions

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

type Tx

type Tx interface {
	Query(query string, args ...interface{}) (pgx.Rows, error)
	Select(dest interface{}, query string, args ...interface{}) error
	Get(dest interface{}, query string, args ...interface{}) error
	GetPrimitive(dest interface{}, query string, args ...interface{}) error
	Exec(query string, args ...interface{}) (pgconn.CommandTag, error)
}

Tx abstract the pg transaction. It is expected to return errors of this package when applicable: 1. ErrNotFound - when a specific row was queried 2. ErrAlreadyExists - on conflicts when adding an entry 3. ErrSerialization - on serialization errors

type TxFunc

type TxFunc func(tx Tx) (interface{}, error)

func Void

func Void(fn func(tx Tx) error) TxFunc

Void wraps a procedure with no return value as a TxFunc

type TxOpt

type TxOpt func(*TxOptions)

func ReadCommitted

func ReadCommitted() TxOpt

func ReadOnly

func ReadOnly() TxOpt

func RepeatableRead

func RepeatableRead() TxOpt

func WithIsolationLevel

func WithIsolationLevel(level pgx.TxIsoLevel) TxOpt

func WithLogger

func WithLogger(logger logging.Logger) TxOpt

type TxOptions

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

func DefaultTxOptions

func DefaultTxOptions() *TxOptions