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, dbPool Database, params params.Database) (uint, bool, error)
• func Ping(ctx context.Context, pool *pgxpool.Pool) error
• func Prefix(prefix string) string
• func Select(d *dbTx, results interface{}, query string, args ...interface{}) error
• func ValidateSchemaUpToDate(ctx context.Context, dbPool Database, 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 ReadOnly() TxOpt
  • func WithIsolationLevel(level pgx.TxIsoLevel) TxOpt
  • func WithLogger(logger logging.Logger) TxOpt
• type TxOptions
  • func DefaultTxOptions(ctx context.Context) *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, dbPool Database, 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 *dbTx, results interface{}, query string, args ...interface{}) error

func ValidateSchemaUpToDate

func ValidateSchemaUpToDate(ctx context.Context, dbPool Database, 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 ReadOnly

func ReadOnly() 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(ctx context.Context) *TxOptions