wpgx

README

Wrapped pgx

Wrapped pgx is a simple wrap on the PostgreSQL driver library pgx. It is used by sqlc, providing telementry for generated code.

Components:

• Pool: a wrapper of pgxpool.Pool. It manages a set of connection pools, including a primary pool and a set of replica pools. We assume replica pools are heterogeneous read-only replicas, meaning some replicas can be a partial copy of the primary database, using logical replication.
• WConn: a connection wrapper, implementing "WGConn".
• WTx: a transaction wrapper, implementing "WGConn".

Documentation

Index

• Constants
• Variables
• type Config
  • func ConfigFromEnv() *Config
  • func ConfigFromEnvPrefix(prefix string) *Config
  • func (c *Config) String() string
  • func (c *Config) Valid() error
• type Pool
  • func NewPool(ctx context.Context, config *Config) (*Pool, error)
  • func (p *Pool) Close()
  • func (p *Pool) MustReplicaPool(name ReplicaName) *pgxpool.Pool
  • func (p *Pool) Ping(ctx context.Context) error
  • func (p *Pool) PingPrimary(ctx context.Context) error
  • func (p *Pool) RawPool() *pgxpool.Pool
  • func (p *Pool) RawPrimaryPool() *pgxpool.Pool
  • func (p *Pool) ReplicaPool(name ReplicaName) (pp *pgxpool.Pool, ok bool)
  • func (p *Pool) ReplicaPools() map[ReplicaName]*pgxpool.Pool
  • func (p *Pool) Transact(ctx context.Context, txOptions pgx.TxOptions, fn TxFunc) (resp interface{}, err error)
  • func (p *Pool) WConn() *WConn
  • func (p *Pool) WQuerierFromReplica(name *ReplicaName) (WQuerier, error)
• type PostExecFunc
• type ReadReplicaConfig
• type ReplicaName
• type TxFunc
• type WConn
  • func (c *WConn) CountIntent(name string)
  • func (c *WConn) PostExec(fn PostExecFunc) error
  • func (c *WConn) WCopyFrom(ctx context.Context, name string, tableName pgx.Identifier, ...) (n int64, err error)
  • func (c *WConn) WExec(ctx context.Context, name string, unprepared string, args ...interface{}) (cmd pgconn.CommandTag, err error)
  • func (c *WConn) WQuery(ctx context.Context, name string, unprepared string, args ...interface{}) (r pgx.Rows, err error)
  • func (c *WConn) WQueryRow(ctx context.Context, name string, unprepared string, args ...interface{}) pgx.Row
• type WCopyFromer
• type WExecer
• type WGConn
• type WQuerier
• type WTx
  • func (t *WTx) Commit(ctx context.Context) error
  • func (t *WTx) CountIntent(name string)
  • func (t *WTx) PostExec(f PostExecFunc) error
  • func (t *WTx) Rollback(ctx context.Context) error
  • func (t *WTx) WCopyFrom(ctx context.Context, name string, tableName pgx.Identifier, ...) (n int64, err error)
  • func (t *WTx) WExec(ctx context.Context, name string, unprepared string, args ...interface{}) (cmd pgconn.CommandTag, err error)
  • func (t *WTx) WQuery(ctx context.Context, name string, unprepared string, args ...interface{}) (rows pgx.Rows, err error)
  • func (t *WTx) WQueryRow(ctx context.Context, name string, unprepared string, args ...interface{}) pgx.Row

Constants

const (
	HighQPSMaxOpenConns = 100
	DefaultEnvPrefix    = "postgres"
	AppNameLengthMax    = 32
)

const (
	// ReservedReplicaNamePrimary is the name of the primary replica.
	ReservedReplicaNamePrimary = "primary"
)

Variables

var (
	// ErrReplicaNotFound is the error when the replica is not found.
	ErrReplicaNotFound = fmt.Errorf("replica not found")
)

Types

type Config

type Config struct {
	Username        string        `default:"postgres"`
	Password        string        `default:"my-secret"`
	Host            string        `default:"localhost"`
	Port            int           `default:"5432"`
	DBName          string        `default:"wpgx_test_db"`
	MaxConns        int32         `default:"100"`
	MinConns        int32         `default:"0"`
	MaxConnLifetime time.Duration `default:"6h"`
	MaxConnIdleTime time.Duration `default:"1m"`
	// BeforeAcquire is a function that is called before acquiring a connection.
	BeforeAcquire func(context.Context, *pgx.Conn) bool `ignored:"true"`
	IsProxy       bool                                  `default:"false"`

	EnablePrometheus bool   `default:"true"`
	EnableTracing    bool   `default:"true"`
	AppName          string `required:"true"`

	// ReplicaConfigPrefixes is a list of replica configuration prefixes. They will
	// be used to create ReadReplicas by using envconfig to parse them.
	ReplicaPrefixes []string `default:""`
	// ReadReplicas is a list of read replicas, parsed from ReplicaNames.
	ReadReplicas []ReadReplicaConfig `ignored:"true"`
}

Config is the configuration for the WPgx. Note: for backward compatibility, connection settings of the primary instance are kept in the root Config, creating a bit code duplication with ReadReplicaConfig.

func ConfigFromEnv

func ConfigFromEnv() *Config

func ConfigFromEnvPrefix

func ConfigFromEnvPrefix(prefix string) *Config

func (*Config) String

func (c *Config) String() string

func (*Config) Valid

func (c *Config) Valid() error

type Pool

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

Pool is the wrapped pgx pool that registers Prometheus.

func NewPool

func NewPool(ctx context.Context, config *Config) (*Pool, error)

NewPool creates a new Pool with the given context and config. When the context is canceled, the pool will be closed.

func (*Pool) Close

func (p *Pool) Close()

Close closes all pools, spawned goroutines, and cancels the context.

func (*Pool) MustReplicaPool

func (p *Pool) MustReplicaPool(name ReplicaName) *pgxpool.Pool

MustReplicaPool returns the replica pool by name, panics if not found.

func (*Pool) Ping

func (p *Pool) Ping(ctx context.Context) error

Ping pings all the instances in the pool, returns the first error encountered.

func (*Pool) PingPrimary

func (p *Pool) PingPrimary(ctx context.Context) error

PingPrimary pings the primary instance in the pool.

func (*Pool) RawPool

func (p *Pool) RawPool() *pgxpool.Pool

RawPool returns the raw primary pgx pool. Deprecated: For backward compatibility only, use RawPrimaryPool instead.

func (*Pool) RawPrimaryPool

func (p *Pool) RawPrimaryPool() *pgxpool.Pool

RawPrimaryPool returns the raw primary pgx pool.

func (*Pool) ReplicaPool

func (p *Pool) ReplicaPool(name ReplicaName) (pp *pgxpool.Pool, ok bool)

ReplicaPool returns the replica pool by name.

func (*Pool) ReplicaPools

func (p *Pool) ReplicaPools() map[ReplicaName]*pgxpool.Pool

RawReplicaPools returns the raw replica pgx pools. NOTE: due to go's lack of constant qualifier, the returned map should be treated as read-only.

func (*Pool) Transact

func (p *Pool) Transact(ctx context.Context, txOptions pgx.TxOptions, fn TxFunc) (resp interface{}, err error)

Transact is a wrapper of pgx.Transaction It acquires a connection from the Pool and starts a transaction with pgx.TxOptions determining the transaction mode. The context will be used when executing the transaction control statements (BEGIN, ROLLBACK, and COMMIT), and when if tracing is enabled, the context with transaction span will be passed down to @p fn.

func (*Pool) WConn

func (p *Pool) WConn() *WConn

WConn returns a wrapped connection for the primary instance.

func (*Pool) WQuerierFromReplica

func (p *Pool) WQuerierFromReplica(name *ReplicaName) (WQuerier, error)

WConnFromReplica returns a wrapped connection for the replica instance by name.

type PostExecFunc

type PostExecFunc = func() error

PostExecFunc is the function that must be ran after a successful CRUD. NOTE: context should have been captured into the function.

type ReadReplicaConfig

type ReadReplicaConfig struct {
	Name            ReplicaName   `required:"true"`
	Username        string        `default:"postgres"`
	Password        string        `default:"my-secret"`
	Host            string        `default:"localhost"`
	Port            int           `default:"5432"`
	DBName          string        `default:"wpgx_test_db"`
	MaxConns        int32         `default:"100"`
	MinConns        int32         `default:"0"`
	MaxConnLifetime time.Duration `default:"6h"`
	MaxConnIdleTime time.Duration `default:"1m"`
	// BeforeAcquire is a function that is called before acquiring a connection.
	BeforeAcquire func(context.Context, *pgx.Conn) bool `ignored:"true"`
	IsProxy       bool                                  `default:"false"`
}

type ReplicaName

type ReplicaName string

ReplicaName is the name of the replica instance.

type TxFunc

type TxFunc = func(ctx context.Context, tx *WTx) (any, error)

TxFunc is the body of a transaction. ctx must be used to generate proper tracing spans. If not, you might see incorrect parallel spans.

type WConn

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

func (*WConn) CountIntent

func (c *WConn) CountIntent(name string)

func (*WConn) PostExec

func (c *WConn) PostExec(fn PostExecFunc) error

func (*WConn) WCopyFrom

func (c *WConn) WCopyFrom(
	ctx context.Context, name string, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (n int64, err error)

func (*WConn) WExec

func (c *WConn) WExec(ctx context.Context, name string, unprepared string, args ...interface{}) (cmd pgconn.CommandTag, err error)

func (*WConn) WQuery

func (c *WConn) WQuery(ctx context.Context, name string, unprepared string, args ...interface{}) (r pgx.Rows, err error)

func (*WConn) WQueryRow

func (c *WConn) WQueryRow(ctx context.Context, name string, unprepared string, args ...interface{}) pgx.Row

type WCopyFromer

type WCopyFromer interface {
	WCopyFrom(
		ctx context.Context, name string, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
}

WCopyFromer is the abstraction of connections that can use PostgreSQL's copyfrom.

type WExecer

type WExecer interface {
	// WExec is used to run a CRUD operation.
	WExec(
		ctx context.Context, name string, unprepared string, args ...interface{}) (pgconn.CommandTag, error)
	// PostExec is used to run a function after a successful CRUD, like invalidating a cache.
	PostExec(f PostExecFunc) error
}

WExecer is the abstraction of connections that can execute mutations.

type WGConn

type WGConn interface {
	WQuerier
	WExecer
	WCopyFromer
}

WGConn is the abstraction over wrapped connections and transactions.

type WQuerier

type WQuerier interface {
	WQuery(
		ctx context.Context, name string, unprepared string, args ...interface{}) (pgx.Rows, error)
	WQueryRow(
		ctx context.Context, name string, unprepared string, args ...interface{}) pgx.Row
	// CountIntent is used to count the number of query intents.
	CountIntent(name string)
}

WQuerier is the abstraction of connections that are read-only.

type WTx

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

WTx is a wrapped pgx.Tx. The main reason is to overwrite the PostExec method that run all of them until the transaction is successfully committed.

func (*WTx) Commit

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

func (*WTx) CountIntent

func (t *WTx) CountIntent(name string)

func (*WTx) PostExec

func (t *WTx) PostExec(f PostExecFunc) error

func (*WTx) Rollback

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

func (*WTx) WCopyFrom

func (t *WTx) WCopyFrom(
	ctx context.Context, name string, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (n int64, err error)

func (*WTx) WExec

func (t *WTx) WExec(ctx context.Context, name string, unprepared string, args ...interface{}) (cmd pgconn.CommandTag, err error)

func (*WTx) WQuery

func (t *WTx) WQuery(ctx context.Context, name string, unprepared string, args ...interface{}) (rows pgx.Rows, err error)

func (*WTx) WQueryRow

func (t *WTx) WQueryRow(ctx context.Context, name string, unprepared string, args ...interface{}) pgx.Row