pgxkit

package module

Go to main page

Versions in this module

v1

v1.4.0

Jan 8, 2026

Changes in this version

+ func Retry[T any](ctx context.Context, fn func(context.Context) (T, error), opts ...RetryOption) (T, error)

v1.3.0

Jan 5, 2026

Changes in this version

+ type RetryOption func(*retryConfig)

+ func WithBackoffMultiplier(m float64) RetryOption

+ func WithBaseDelay(d time.Duration) RetryOption

+ func WithMaxDelay(d time.Duration) RetryOption

+ func WithMaxRetries(n int) RetryOption

v1.2.0

Jan 4, 2026

Changes in this version

+ type ConnectOption func(*connectConfig)

+ func WithAfterOperation(fn HookFunc) ConnectOption

+ func WithAfterTransaction(fn HookFunc) ConnectOption

+ func WithBeforeOperation(fn HookFunc) ConnectOption

+ func WithBeforeTransaction(fn HookFunc) ConnectOption

+ func WithMaxConnIdleTime(d time.Duration) ConnectOption

+ func WithMaxConnLifetime(d time.Duration) ConnectOption

+ func WithMaxConns(n int32) ConnectOption

+ func WithMinConns(n int32) ConnectOption

+ func WithOnAcquire(fn func(context.Context, *pgx.Conn) error) ConnectOption

+ func WithOnConnect(fn func(*pgx.Conn) error) ConnectOption

+ func WithOnDisconnect(fn func(*pgx.Conn)) ConnectOption

+ func WithOnRelease(fn func(*pgx.Conn)) ConnectOption

+ func WithOnShutdown(fn HookFunc) ConnectOption

+ func WithReadMaxConns(n int32) ConnectOption

+ func WithReadMinConns(n int32) ConnectOption

+ func WithWriteMaxConns(n int32) ConnectOption

+ func WithWriteMinConns(n int32) ConnectOption

type DB

+ func (db *DB) ReadPool() *pgxpool.Pool

+ func (db *DB) WritePool() *pgxpool.Pool

v1.1.0

Jul 16, 2025

Changes in this version

+ func CleanupGolden(testName string) error

+ func FromPgxBoolToBool(b pgtype.Bool) bool

+ func FromPgxDate(d pgtype.Date) *time.Time

+ func FromPgxFloat4(f pgtype.Float4) *float32

+ func FromPgxFloat8(f pgtype.Float8) *float64

+ func FromPgxInt2(i pgtype.Int2) *int16

+ func FromPgxInt4ToInt(i pgtype.Int4) *int

+ func FromPgxInt8Array(a pgtype.Array[pgtype.Int8]) []int64

+ func FromPgxNumeric(n pgtype.Numeric) *float64

+ func FromPgxTextArray(a pgtype.Array[pgtype.Text]) []string

+ func FromPgxTime(t pgtype.Time) *time.Time

+ func FromPgxTimestamp(t pgtype.Timestamp) *time.Time

+ func FromPgxUUIDToPtr(pgxID pgtype.UUID) *uuid.UUID

+ func IsRetryableError(err error) bool

+ func RetryOperation(ctx context.Context, config *RetryConfig, ...) error

+ func ToPgxBoolFromBool(b bool) pgtype.Bool

+ func ToPgxDate(t *time.Time) pgtype.Date

+ func ToPgxFloat4(f *float32) pgtype.Float4

+ func ToPgxFloat8(f *float64) pgtype.Float8

+ func ToPgxInt2(i *int16) pgtype.Int2

+ func ToPgxInt4(i *int32) pgtype.Int4

+ func ToPgxInt8Array(s []int64) pgtype.Array[pgtype.Int8]

+ func ToPgxNumeric(f *float64) pgtype.Numeric

+ func ToPgxTextArray(s []string) pgtype.Array[pgtype.Text]

+ func ToPgxTextFromString(s string) pgtype.Text

+ func ToPgxTime(t *time.Time) pgtype.Time

+ func ToPgxTimestamp(t *time.Time) pgtype.Timestamp

+ func ToPgxUUIDFromPtr(id *uuid.UUID) pgtype.UUID

type ConnectionHooks

+ func (ch *ConnectionHooks) ConfigurePool(config *pgxpool.Config)

+ type DB struct

+ func NewDB() *DB

+ func NewDBWithPool(pool *pgxpool.Pool) *DB

+ func NewReadWriteDB(readPool, writePool *pgxpool.Pool) *DB

+ func (db *DB) AddConnectionHook(hookType string, hookFunc interface{}) error

+ func (db *DB) AddHook(hookType HookType, hookFunc HookFunc) *DB

+ func (db *DB) AssertGolden(t *testing.T, testName string)

+ func (db *DB) BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)

+ func (db *DB) BeginTxWithRetry(ctx context.Context, config *RetryConfig, txOptions pgx.TxOptions) (pgx.Tx, error)

+ func (db *DB) Connect(ctx context.Context, dsn string) error

+ func (db *DB) ConnectReadWrite(ctx context.Context, readDSN, writeDSN string) error

+ func (db *DB) Exec(ctx context.Context, sql string, args ...interface{}) (pgconn.CommandTag, error)

+ func (db *DB) ExecWithRetry(ctx context.Context, config *RetryConfig, sql string, args ...interface{}) (pgconn.CommandTag, error)

+ func (db *DB) HealthCheck(ctx context.Context) error

+ func (db *DB) IsReady(ctx context.Context) bool

+ func (db *DB) Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)

+ func (db *DB) QueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row

+ func (db *DB) QueryRowWithRetry(ctx context.Context, config *RetryConfig, sql string, args ...interface{}) pgx.Row

+ func (db *DB) QueryWithRetry(ctx context.Context, config *RetryConfig, sql string, args ...interface{}) (pgx.Rows, error)

+ func (db *DB) ReadQuery(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)

+ func (db *DB) ReadQueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row

+ func (db *DB) ReadQueryRowWithRetry(ctx context.Context, config *RetryConfig, sql string, args ...interface{}) pgx.Row

+ func (db *DB) ReadQueryWithRetry(ctx context.Context, config *RetryConfig, sql string, args ...interface{}) (pgx.Rows, error)

+ func (db *DB) ReadStats() *pgxpool.Stat

+ func (db *DB) Shutdown(ctx context.Context) error

+ func (db *DB) Stats() *pgxpool.Stat

+ func (db *DB) WriteStats() *pgxpool.Stat

+ type DBConfig struct

+ MaxConnIdleTime time.Duration

+ MaxConnLifetime time.Duration

+ MaxConns int32

+ MinConns int32

+ type HookFunc func(ctx context.Context, sql string, args []interface{}, operationErr error) error

+ type HookType int

+ const AfterOperation

+ const AfterTransaction

+ const BeforeOperation

+ const BeforeTransaction

+ const OnShutdown

+ type QueryPlan struct

+ ExecutionMS float64

+ Plan []map[string]interface{}

+ PlanningMS float64

+ Query int

+ SQL string

+ type TestDB struct

+ func NewTestDB() *TestDB

+ func RequireDB(t *testing.T) *TestDB

+ func (tdb *TestDB) Clean() error

+ func (tdb *TestDB) EnableGolden(testName string) *DB

+ func (tdb *TestDB) Setup() error

v1.0.1

Jul 15, 2025

v1.0.0

Jul 15, 2025

Changes in this version

+ func CleanupTestData[T Querier](conn *Connection[T], sqlStatements ...string)

+ func FromPgxBool(b pgtype.Bool) *bool

+ func FromPgxInt4(val pgtype.Int4) *int

+ func FromPgxInt8(i pgtype.Int8) *int64

+ func FromPgxNumericPtr(val pgtype.Numeric) *float64

+ func FromPgxText(t pgtype.Text) *string

+ func FromPgxTextToString(val pgtype.Text) string

+ func FromPgxTimestamptz(ts pgtype.Timestamptz) time.Time

+ func FromPgxTimestamptzPtr(val pgtype.Timestamptz) *time.Time

+ func FromPgxUUID(pgxID pgtype.UUID) uuid.UUID

+ func GetDSN() string

+ func ToPgxBool(b *bool) pgtype.Bool

+ func ToPgxInt4FromInt(val *int) pgtype.Int4

+ func ToPgxInt8(i *int64) pgtype.Int8

+ func ToPgxNumericFromFloat64Ptr(val *float64) pgtype.Numeric

+ func ToPgxText(s *string) pgtype.Text

+ func ToPgxTimestamptz(t *time.Time) pgtype.Timestamptz

+ func ToPgxUUID(id uuid.UUID) pgtype.UUID

+ func WithTimeoutAndRetry[T any](ctx context.Context, timeout time.Duration, retryConfig *RetryConfig, ...) (T, error)

+ func WithTimeout[T any](ctx context.Context, timeout time.Duration, ...) (T, error)

+ type Config struct

+ Hooks *ConnectionHooks

+ MaxConnLifetime time.Duration

+ MaxConns int32

+ MinConns int32

+ OnConnect func(*pgx.Conn) error

+ OnDisconnect func(*pgx.Conn)

+ SearchPath string

+ type Connection struct

+ func GetTestConnection[T Querier](newQueriesFunc func(*pgxpool.Pool) T) *Connection[T]

+ func NewConnectionWithConfig[T Querier](ctx context.Context, dsn string, newQueriesFunc func(*pgxpool.Pool) T, ...) (*Connection[T], error)

+ func NewConnectionWithHooks[T Querier](ctx context.Context, dsn string, newQueriesFunc func(*pgxpool.Pool) T, ...) (*Connection[T], error)

+ func NewConnectionWithLoggingHooks[T Querier](ctx context.Context, dsn string, newQueriesFunc func(*pgxpool.Pool) T, ...) (*Connection[T], error)

+ func NewConnectionWithValidationHooks[T Querier](ctx context.Context, dsn string, newQueriesFunc func(*pgxpool.Pool) T) (*Connection[T], error)

+ func NewConnection[T Querier](ctx context.Context, dsn string, newQueriesFunc func(*pgxpool.Pool) T) (*Connection[T], error)

+ func RequireTestDB[T Querier](t TestingT, newQueriesFunc func(*pgxpool.Pool) T) *Connection[T]

+ func (c *Connection[T]) AddHook(hook *ConnectionHooks) *Connection[T]

+ func (c *Connection[T]) BeginTransaction(ctx context.Context) (pgx.Tx, T, error)

+ func (c *Connection[T]) Close()

+ func (c *Connection[T]) GetDB() *pgxpool.Pool

+ func (c *Connection[T]) GetHooks() *ConnectionHooks

+ func (c *Connection[T]) HealthCheck(ctx context.Context) error

+ func (c *Connection[T]) IsReady(ctx context.Context) bool

+ func (c *Connection[T]) Queries() T

+ func (c *Connection[T]) Stats() *pgxpool.Stat

+ func (c *Connection[T]) WithHooks(hooks *ConnectionHooks) *Connection[T]

+ func (c *Connection[T]) WithLogging(logger Logger) *LoggingConnection[T]

+ func (c *Connection[T]) WithMetrics(metrics MetricsCollector) *Connection[T]

+ func (c *Connection[T]) WithRetry(config *RetryConfig) *RetryableConnection[T]

+ func (c *Connection[T]) WithTransaction(ctx context.Context, fn TransactionFunc[T]) error

+ type ConnectionHooks struct

+ func CombineHooks(hooksList ...*ConnectionHooks) *ConnectionHooks

+ func LoggingHook(logger Logger) *ConnectionHooks

+ func MetricsHook(metrics MetricsCollector) *ConnectionHooks

+ func NewConnectionHooks() *ConnectionHooks

+ func SetupHook(setupSQL string) *ConnectionHooks

+ func ValidationHook() *ConnectionHooks

+ func (h *ConnectionHooks) AddOnAcquire(fn func(context.Context, *pgx.Conn) error)

+ func (h *ConnectionHooks) AddOnConnect(fn func(*pgx.Conn) error)

+ func (h *ConnectionHooks) AddOnDisconnect(fn func(*pgx.Conn))

+ func (h *ConnectionHooks) AddOnRelease(fn func(*pgx.Conn))

+ func (h *ConnectionHooks) ExecuteOnAcquire(ctx context.Context, conn *pgx.Conn) error

+ func (h *ConnectionHooks) ExecuteOnConnect(conn *pgx.Conn) error

+ func (h *ConnectionHooks) ExecuteOnDisconnect(conn *pgx.Conn)

+ func (h *ConnectionHooks) ExecuteOnRelease(conn *pgx.Conn)

+ type DatabaseError struct

+ Entity string

+ Err error

+ Operation string

+ func NewDatabaseError(entity, operation string, err error) *DatabaseError

+ func (e *DatabaseError) Error() string

+ func (e *DatabaseError) Unwrap() error

+ type DefaultLogger struct

+ func NewDefaultLogger(minLevel LogLevel) *DefaultLogger

+ func (l *DefaultLogger) Log(ctx context.Context, level LogLevel, msg string, data map[string]interface{})

+ type LogLevel int

+ const LogLevelDebug

+ const LogLevelError

+ const LogLevelInfo

+ const LogLevelNone

+ const LogLevelTrace

+ const LogLevelWarn

+ func (l LogLevel) String() string

+ type Logger interface

+ Log func(ctx context.Context, level LogLevel, msg string, data map[string]interface{})

+ type LoggingConfig struct

+ LogConnections bool

+ LogLevel LogLevel

+ LogSlowQueries bool

+ LogTransactions bool

+ Logger Logger

+ SlowQueryThreshold time.Duration

+ func DefaultLoggingConfig() *LoggingConfig

+ type LoggingConnection struct

+ func NewConnectionWithConfigAndLogging[T Querier](ctx context.Context, dsn string, newQueriesFunc func(*pgxpool.Pool) T, ...) (*LoggingConnection[T], error)

+ func NewConnectionWithLogging[T Querier](ctx context.Context, dsn string, newQueriesFunc func(*pgxpool.Pool) T, ...) (*LoggingConnection[T], error)

+ func (lc *LoggingConnection[T]) WithTransaction(ctx context.Context, fn TransactionFunc[T]) error

+ type MetricsCollector interface

+ RecordConnectionAcquired func(duration time.Duration)

+ RecordConnectionReleased func(duration time.Duration)

+ RecordQueryExecuted func(queryName string, duration time.Duration, err error)

+ RecordTransactionCommitted func(duration time.Duration)

+ RecordTransactionRolledBack func(duration time.Duration)

+ RecordTransactionStarted func()

+ type NotFoundError struct

+ Entity string

+ Identifier interface{}

+ func NewNotFoundError(entity string, identifier interface{}) *NotFoundError

+ func (e *NotFoundError) Error() string

+ type Querier interface

+ WithTx func(tx pgx.Tx) Querier

+ type QueryLogger struct

+ func NewQueryLogger[T Querier](queries T, logger Logger) *QueryLogger[T]

+ func (ql *QueryLogger[T]) LogQuery(ctx context.Context, queryName string, fn func() error) error

+ type ReadWriteConnection struct

+ func NewReadWriteConnectionWithConfig[T Querier](ctx context.Context, readDSN, writeDSN string, ...) (*ReadWriteConnection[T], error)

+ func NewReadWriteConnection[T Querier](ctx context.Context, readDSN, writeDSN string, ...) (*ReadWriteConnection[T], error)

+ func (rw *ReadWriteConnection[T]) BeginTransaction(ctx context.Context) (pgx.Tx, T, error)

+ func (rw *ReadWriteConnection[T]) Close()

+ func (rw *ReadWriteConnection[T]) HealthCheck(ctx context.Context) error

+ func (rw *ReadWriteConnection[T]) IsReady(ctx context.Context) bool

+ func (rw *ReadWriteConnection[T]) ReadDB() *pgxpool.Pool

+ func (rw *ReadWriteConnection[T]) ReadQueries() T

+ func (rw *ReadWriteConnection[T]) ReadStats() *pgxpool.Stat

+ func (rw *ReadWriteConnection[T]) WithHooks(hooks *ConnectionHooks) *ReadWriteConnection[T]

+ func (rw *ReadWriteConnection[T]) WithMetrics(metrics MetricsCollector) *ReadWriteConnection[T]

+ func (rw *ReadWriteConnection[T]) WithRetry(config *RetryConfig) *RetryableReadWriteConnection[T]

+ func (rw *ReadWriteConnection[T]) WithTransaction(ctx context.Context, fn TransactionFunc[T]) error

+ func (rw *ReadWriteConnection[T]) WriteDB() *pgxpool.Pool

+ func (rw *ReadWriteConnection[T]) WriteQueries() T

+ func (rw *ReadWriteConnection[T]) WriteStats() *pgxpool.Stat

+ type RetryConfig struct

+ BaseDelay time.Duration

+ MaxDelay time.Duration

+ MaxRetries int

+ Multiplier float64

+ func DefaultRetryConfig() *RetryConfig

+ type RetryableConnection struct

+ func (rc *RetryableConnection[T]) WithRetryableTransaction(ctx context.Context, fn TransactionFunc[T]) error

+ type RetryableReadWriteConnection struct

+ func (rrc *RetryableReadWriteConnection[T]) WithRetryableTransaction(ctx context.Context, fn TransactionFunc[T]) error

+ type SlowQueryLogger struct

+ func NewSlowQueryLogger(logger Logger, threshold time.Duration) *SlowQueryLogger

+ func (sql *SlowQueryLogger) LogIfSlow(ctx context.Context, queryName string, duration time.Duration, err error)

+ type TestingT interface

+ Logf func(format string, args ...interface{})

+ Skip func(args ...interface{})

+ type TransactionFunc func(ctx context.Context, tx T) error

+ type ValidationError struct

+ Entity string

+ Err error

+ Field string

+ Operation string

+ Reason string

+ func NewValidationError(entity, operation, field, reason string, err error) *ValidationError

+ func (e *ValidationError) Error() string

+ func (e *ValidationError) Unwrap() error

Other modules containing this package

github.com/nhalm/pgxkit/v2

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL