pg

Documentation

Overview

Package pg is a wrapper around the pgx PostgreSQL driver, providing a higher-level client with features for connection pooling, tracing, and metrics.

Index

• Constants
• func NoRollback(err error) error
• type AdvisoryLock
• type Client
  • func NewClient(options ...Option) (*Client, error)
  • func (c *Client) Close()
  • func (c *Client) RefreshTypes(ctx context.Context) error
  • func (c *Client) WithAdvisoryLock(ctx context.Context, id AdvisoryLock, f func(Querier) error) error
  • func (c *Client) WithConn(ctx context.Context, exec ExecFunc[Querier]) error
  • func (c *Client) WithTx(ctx context.Context, exec ExecFunc[Tx]) error
• type ExecFunc
• type NoRollbackError
  • func (e *NoRollbackError) Error() string
  • func (e *NoRollbackError) Unwrap() error
• type Option
  • func WithAddr(addr string) Option
  • func WithApplicationName(name string) Option
  • func WithDatabase(database string) Option
  • func WithDebug() Option
  • func WithHealthCheckPeriod(d time.Duration) Option
  • func WithLogger(l *log.Logger) Option
  • func WithMaxConnIdleTime(d time.Duration) Option
  • func WithMaxConnLifetime(d time.Duration) Option
  • func WithMaxConnLifetimeJitter(d time.Duration) Option
  • func WithMinPoolSize(i int32) Option
  • func WithPassword(password string) Option
  • func WithPoolSize(i int32) Option
  • func WithRegisterer(r prometheus.Registerer) Option
  • func WithTLS(certs []*x509.Certificate) Option
  • func WithTracerProvider(tp trace.TracerProvider) Option
  • func WithUnsecureTLS() Option
  • func WithUser(user string) Option
• type Querier
• type Tx

Constants

const (

	// BatchSizeKey represents the batch size.
	BatchSizeKey = attribute.Key("db.operation.batch.size")

	// PrepareStmtNameKey represents the prepared statement name.
	PrepareStmtNameKey = attribute.Key("pgx.prepare_stmt.name")

	// RowsAffectedKey represents the number of rows affected.
	RowsAffectedKey = attribute.Key("pgx.rows_affected")

	// SQLStateKey represents PostgreSQL error code,
	// see https://www.postgresql.org/docs/current/errcodes-appendix.html.
	SQLStateKey = attribute.Key("db.response.status_code")
)

const (
	BaseAdvisoryLockId uint32 = 42
)

Functions

func NoRollback

func NoRollback(err error) error

NoRollback wraps err so that WithTx commits the transaction instead of rolling back. The inner error is still returned to the caller after the commit.

A nil err is returned as-is (no wrapping).

Types

type AdvisoryLock

type AdvisoryLock = uint32

type Client

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

Client provides a PostgreSQL client with a connection pool, logging, tracing, and Prometheus metrics registration.

func NewClient

func NewClient(options ...Option) (*Client, error)

NewClient creates a new database client with customizable options for logging, tracing, TLS, and Prometheus metrics.

Example:

client, err := pg.NewClient(
    pg.WithAddr("db.example.com:5432"),
    pg.WithUser("dbuser"),
    pg.WithPassword("password"),
)
if err != nil {
    panic(err)
}

func (*Client) Close

func (c *Client) Close()

Close closes the client's connection pool, releasing all resources.

func (*Client) RefreshTypes

func (c *Client) RefreshTypes(ctx context.Context) error

func (*Client) WithAdvisoryLock

func (c *Client) WithAdvisoryLock(
	ctx context.Context,
	id AdvisoryLock,
	f func(Querier) error,
) error

func (*Client) WithConn

func (c *Client) WithConn(
	ctx context.Context,
	exec ExecFunc[Querier],
) error

WithConn executes the given ExecFunc with a database connection from the pool.

Example:

err := client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
    _, err := conn.Exec(ctx, "SELECT * FROM users")
    return err
})

If tracing is enabled, this method creates a span named "WithConn" and logs any errors.

func (*Client) WithTx

func (c *Client) WithTx(
	ctx context.Context,
	exec ExecFunc[Tx],
) error

WithTx executes the given ExecFunc within a new transaction. This method acquires a connection, begins a transaction, and executes exec within it. If exec returns an error, the transaction is rolled back; otherwise, it commits.

Use Tx.Savepoint inside the callback to create savepoints within the transaction.

Example:

err := client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
    if _, err := tx.Exec(ctx, "DELETE FROM users WHERE id = $1", id); err != nil {
        return err
    }

    // Savepoint failure does not roll back the DELETE above.
    // The callback receives a Tx, so savepoints can be nested.
    if err := tx.Savepoint(ctx, func(ctx context.Context, inner pg.Tx) error {
        _, err := inner.Exec(ctx, "INSERT INTO audit_log (...) VALUES (...)")
        return err
    }); err != nil {
        log.Warn("audit failed, continuing", "err", err)
    }

    return nil
})

If tracing is enabled, this method creates a span named "WithTx" and logs any errors.

type ExecFunc

type ExecFunc[Q Querier] func(context.Context, Q) error

type NoRollbackError

type NoRollbackError struct {
	Err error
}

NoRollbackError wraps an error to signal that WithTx should commit the transaction instead of rolling back, while still propagating the inner error to the caller.

func (*NoRollbackError) Error

func (e *NoRollbackError) Error() string

func (*NoRollbackError) Unwrap

func (e *NoRollbackError) Unwrap() error

type Option

type Option func(c *Client)

Option is a function that configures the Client during initialization.

func WithAddr

func WithAddr(addr string) Option

WithAddr specifies the database address in "host:port" format.

func WithApplicationName

func WithApplicationName(name string) Option

WithApplicationName sets the PostgreSQL application_name runtime parameter on every connection in the pool. This value is exposed by the server in pg_stat_activity, pg_stat_statements, log lines, and pg_locks, making it the primary handle for identifying the source of a query during incident response.

PostgreSQL truncates application_name to NAMEDATALEN-1 (63 bytes by default) silently, so prefer short, stable identifiers such as the service name. An empty string leaves the runtime parameter unset (which falls back to whatever the libpq env, e.g. PGAPPNAME, or the server defaults choose).

func WithDatabase

func WithDatabase(database string) Option

WithDatabase specifies the database to connect to.

func WithDebug

func WithDebug() Option

WithDebug enables debug logging for the client.

func WithHealthCheckPeriod

func WithHealthCheckPeriod(d time.Duration) Option

WithHealthCheckPeriod sets how often the pool checks the health of its idle connections and enforces MinConns/MaxConnIdleTime/ MaxConnLifetime. It maps to pgxpool.Config.HealthCheckPeriod. A zero value leaves the pgx default (1 minute) in place.

Lowering the period makes the pool react faster to dropped connections (for example after a failover) and refill below MinConns more promptly, at the cost of slightly more background work.

func WithLogger

func WithLogger(l *log.Logger) Option

WithLogger sets a custom logger.

func WithMaxConnIdleTime

func WithMaxConnIdleTime(d time.Duration) Option

WithMaxConnIdleTime sets how long a connection can be idle before it is eligible to be destroyed. It maps to pgxpool.Config.MaxConnIdleTime. A zero value leaves the pgx default (30 minutes) in place.

Increasing this value reduces connection churn for bursty workloads, at the cost of holding idle connections longer.

func WithMaxConnLifetime

func WithMaxConnLifetime(d time.Duration) Option

WithMaxConnLifetime sets the maximum lifetime of a connection before it is recycled. It maps to pgxpool.Config.MaxConnLifetime. A zero value leaves the pgx default (1 hour) in place.

Increasing this value reduces connection churn for long-lived processes, at the cost of keeping the same backend connection open for longer.

func WithMaxConnLifetimeJitter

func WithMaxConnLifetimeJitter(d time.Duration) Option

WithMaxConnLifetimeJitter sets the duration of randomness added to each connection's MaxConnLifetime, smearing recycle events across time. It maps to pgxpool.Config.MaxConnLifetimeJitter.

Without jitter, connections opened around the same time (for example after a deploy) tend to be recycled together, producing synchronized reconnect storms that show up as latency spikes on pgxpool_acquire_duration_seconds. A non-zero jitter spreads those recycles out.

When unset, the client defaults to 5 minutes. Pass a zero value explicitly via this option to disable jitter.

func WithMinPoolSize

func WithMinPoolSize(i int32) Option

WithMinPoolSize sets the minimum number of connections the pool keeps warm. It maps to pgxpool.Config.MinConns. Keeping a non-zero floor avoids paying the TCP + TLS + PostgreSQL startup handshake cost on the first acquire after an idle period, which otherwise shows up as tail latency on pgxpool_acquire_duration_seconds.

When unset, the client defaults to 1 (historical behaviour).

func WithPassword

func WithPassword(password string) Option

WithPassword sets the database password.

func WithPoolSize

func WithPoolSize(i int32) Option

WithPoolSize sets the maximum number of connections the pool will open. It maps to pgxpool.Config.MaxConns.

func WithRegisterer

func WithRegisterer(r prometheus.Registerer) Option

WithRegisterer sets a custom Prometheus registerer for metrics.

func WithTLS

func WithTLS(certs []*x509.Certificate) Option

WithTLS configures TLS using the provided certificates for secure connections.

The returned config enables a TLS client session cache so that reconnects (after MaxConnLifetime recycles, failovers, or pool refills) can resume a previous session and skip the full handshake. This noticeably reduces tail latency on pgxpool_acquire_duration_seconds for deployments where the database is reached over WAN or any link where the round-trip cost dominates connection construction.

func WithTracerProvider

func WithTracerProvider(tp trace.TracerProvider) Option

WithTracerProvider configures OpenTelemetry tracing with the provided tracer provider.

func WithUnsecureTLS

func WithUnsecureTLS() Option

WithUnsecureTLS enables TLS without verifying the server's certificate chain or hostname. Intended for development against self-signed databases; do not use in production.

func WithUser

func WithUser(user string) Option

WithUser sets the database user.

type Querier

type Querier interface {
	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)

	Query(context.Context, string, ...any) (pgx.Rows, error)
	QueryRow(context.Context, string, ...any) pgx.Row

	CopyFrom(context.Context, pgx.Identifier, []string, pgx.CopyFromSource) (int64, error)
	SendBatch(context.Context, *pgx.Batch) pgx.BatchResults
}

Querier represents something you can run SQL queries against.

type Tx

type Tx interface {
	Querier

	Savepoint(context.Context, ExecFunc[Tx]) error
}

Tx represents an active database transaction. It extends Querier with the ability to create savepoints. The callback receives a Tx so savepoints can be nested arbitrarily.