pgxdriver

Documentation

Overview

Package pgxdriver provides a robust PostgreSQL client built on pgx/v5, featuring connection retries with exponential backoff and jitter, integrated SQL query building via squirrel, and structured logging.

Index

• Variables
• func BatchInsert(ctx context.Context, qe QueryExecuter, sql string, rows [][]any) error
• func BulkInsert(ctx context.Context, qe QueryExecuter, tableName any, columns []string, ...) (int64, error)
• type Option
  • func BaseRetryDelay(delay time.Duration) Option
  • func MaxConnAttempts(attempts int) Option
  • func MaxPoolSize(size int32) Option
  • func MaxRetryDelay(delay time.Duration) Option
• type Postgres
  • func New(dsn string, logger logger.Logger, opts ...Option) (*Postgres, error)
  • func (p *Postgres) Close()
  • func (p *Postgres) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, ...) (int64, error)
  • func (p *Postgres) Delete(from string) squirrel.DeleteBuilder
  • func (p *Postgres) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
  • func (p *Postgres) Insert(into string) squirrel.InsertBuilder
  • func (p *Postgres) Ping(ctx context.Context) error
  • func (p *Postgres) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
  • func (p *Postgres) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
  • func (p *Postgres) Select(columns ...string) squirrel.SelectBuilder
  • func (p *Postgres) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
  • func (p *Postgres) Update(table string) squirrel.UpdateBuilder
• type QueryExecuter
• type TxQueryExecuter
  • func (t *TxQueryExecuter) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, ...) (int64, error)
  • func (t *TxQueryExecuter) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
  • func (t *TxQueryExecuter) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
  • func (t *TxQueryExecuter) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
  • func (t *TxQueryExecuter) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults

Variables

var (
	// ErrInvalidMaxPoolSize is returned when MaxPoolSize <= 0.
	ErrInvalidMaxPoolSize = errors.New("invalid maxPoolSize: must be > 0")
	// ErrInvalidConnAttempts is returned when MaxConnAttempts <= 0.
	ErrInvalidConnAttempts = errors.New("invalid connAttempts: must be > 0")
	// ErrInvalidBaseRetryDelay is returned when BaseRetryDelay <= 0.
	ErrInvalidBaseRetryDelay = errors.New("invalid base retry delay: must be > 0")
	// ErrInvalidMaxRetryDelay is returned when MaxRetryDelay <= 0.
	ErrInvalidMaxRetryDelay = errors.New("invalid max retry delay: must be > 0")
	// ErrBaseExceedsMaxDelay is returned when BaseRetryDelay > MaxRetryDelay.
	ErrBaseExceedsMaxDelay = errors.New("baseRetryDelay cannot exceed maxRetryDelay")
)

var ErrInvalidTableName = errors.New("invalid table name type")

ErrInvalidTableName is returned when tableName is not string, []string, or pgx.Identifier.

Functions

func BatchInsert

func BatchInsert(ctx context.Context, qe QueryExecuter, sql string, rows [][]any) error

BatchInsert executes the same SQL statement multiple times with different parameters using pgx's batch protocol for improved performance over individual Exec calls. It accepts a slice of parameter rows and queues each as a separate batch operation. The function works with any QueryExecuter (e.g., *Postgres or *TxQueryExecuter), enabling use both inside and outside of transactions. Returns an error if any statement in the batch fails.

func BulkInsert

func BulkInsert(ctx context.Context, qe QueryExecuter, tableName any, columns []string, data [][]any) (int64, error)

BulkInsert performs a high-performance bulk insert into a PostgreSQL table using the COPY FROM protocol. It accepts the table name as a string, []string, or pgx.Identifier, column names, and a 2D slice of row data. The function works with any implementation of QueryExecuter (e.g., *Postgres or *TxQueryExecuter), making it usable both inside and outside of transactions. Returns the number of rows inserted and an error, if any occurred.

Types

type Option

type Option func(*Postgres)

Option represents a functional configuration option for the Postgres client. Use Option functions like MaxPoolSize or BaseRetryDelay to customize client behavior.

func BaseRetryDelay

func BaseRetryDelay(delay time.Duration) Option

BaseRetryDelay sets the initial delay for the exponential backoff retry logic when connecting to the database. The value must be greater than zero.

func MaxConnAttempts

func MaxConnAttempts(attempts int) Option

MaxConnAttempts sets the maximum number of attempts to establish a database connection during client initialization. The value must be greater than zero.

func MaxPoolSize

func MaxPoolSize(size int32) Option

MaxPoolSize sets the maximum number of concurrent connections in the connection pool. The size must be greater than zero.

func MaxRetryDelay

func MaxRetryDelay(delay time.Duration) Option

MaxRetryDelay sets the upper bound for retry delays during connection attempts. The backoff delay will never exceed this value. The value must be greater than zero and greater than or equal to BaseRetryDelay.

type Postgres

type Postgres struct {
	Builder squirrel.StatementBuilderType
	Pool    *pgxpool.Pool
	// contains filtered or unexported fields
}

Postgres represents a PostgreSQL client with a connection pool and SQL builder. It wraps pgxpool.Pool and provides methods for querying, building SQL statements, and managing the lifecycle of the database connection.

func New

func New(dsn string, logger logger.Logger, opts ...Option) (*Postgres, error)

New creates and initializes a new Postgres client. It parses the provided DSN, applies configuration options, and attempts to establish a connection pool with exponential backoff and jitter. Returns an error if validation fails, the DSN is invalid, or all connection attempts are exhausted.

func (*Postgres) Close

func (p *Postgres) Close()

Close gracefully shuts down the connection pool and logs the shutdown process. It is safe to call Close multiple times.

func (*Postgres) CopyFrom

func (p *Postgres) CopyFrom(
	ctx context.Context,
	tableName pgx.Identifier,
	columnNames []string,
	rowSrc pgx.CopyFromSource,
) (int64, error)

CopyFrom performs a high-performance bulk insert using PostgreSQL's COPY FROM protocol. Delegates to the underlying pgxpool.Pool.

func (*Postgres) Delete

func (p *Postgres) Delete(from string) squirrel.DeleteBuilder

Delete starts a new DELETE query using the embedded squirrel builder. The `from` parameter specifies the table to delete from.

func (*Postgres) Exec

func (p *Postgres) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

Exec executes a non-query SQL statement (e.g., INSERT, UPDATE, DELETE). Delegates to the underlying pgxpool.Pool.

func (*Postgres) Insert

func (p *Postgres) Insert(into string) squirrel.InsertBuilder

Insert starts a new INSERT query using the embedded squirrel builder. The `into` parameter specifies the target table name.

func (*Postgres) Ping

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

Ping verifies the database connection by sending a lightweight ping request. It returns an error if the connection is not alive.

func (*Postgres) Query

func (p *Postgres) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)

Query executes a query that returns rows, such as a SELECT. Delegates to the underlying pgxpool.Pool.

func (*Postgres) QueryRow

func (p *Postgres) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row

QueryRow executes a query expected to return at most one row. Delegates to the underlying pgxpool.Pool.

func (*Postgres) Select

func (p *Postgres) Select(columns ...string) squirrel.SelectBuilder

Select starts a new SELECT query using the embedded squirrel builder. The returned builder supports chaining methods like From, Where, OrderBy, etc.

func (*Postgres) SendBatch

func (p *Postgres) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults

SendBatch sends a batch of queries to the server using pgx's batch protocol. Delegates to the underlying pgxpool.Pool.

func (*Postgres) Update

func (p *Postgres) Update(table string) squirrel.UpdateBuilder

Update starts a new UPDATE query using the embedded squirrel builder. The `table` parameter specifies the table to update.

type QueryExecuter

type QueryExecuter interface {
	// Query executes a query that returns rows, such as a SELECT.
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)

	// QueryRow executes a query that is expected to return at most one row.
	// It is safe to call Scan on the returned Row even if no row is found.
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row

	// Exec executes a query that does not return rows, such as INSERT, UPDATE, or DELETE.
	// Returns the command tag and any error.
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

	// SendBatch sends a batch of queries to the server in a single round trip.
	SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults

	// CopyFrom performs a PostgreSQL COPY FROM operation for high-performance bulk inserts.
	// Returns the number of rows copied.
	CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
}

QueryExecuter defines a unified interface for executing SQL queries and commands. It is implemented by both the main Postgres client and transaction wrappers, enabling seamless use of the same logic in and outside of transactions.

type TxQueryExecuter

type TxQueryExecuter struct {
	Tx pgx.Tx
}

TxQueryExecuter wraps a pgx.Tx to satisfy the QueryExecuter interface, allowing transactional and non-transactional code to share the same execution path.

func (*TxQueryExecuter) CopyFrom

func (t *TxQueryExecuter) CopyFrom(
	ctx context.Context,
	tableName pgx.Identifier,
	columnNames []string,
	rowSrc pgx.CopyFromSource,
) (int64, error)

CopyFrom performs a COPY FROM operation within a transaction.

func (*TxQueryExecuter) Exec

func (t *TxQueryExecuter) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

Exec executes a non-query statement within a transaction.

func (*TxQueryExecuter) Query

func (t *TxQueryExecuter) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)

Query executes a query within a transaction.

func (*TxQueryExecuter) QueryRow

func (t *TxQueryExecuter) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row

QueryRow executes a single-row query within a transaction.

func (*TxQueryExecuter) SendBatch

func (t *TxQueryExecuter) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults

SendBatch sends a batch of queries within a transaction.