pgfx

package module
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 25, 2026 License: MIT Imports: 17 Imported by: 0

README

pgfx

CI Go Reference

PostgreSQL connections, queries, transactions, tracing, and Uber Fx lifecycle management on top of pgx.

Installation

go get github.com/uchaloop/pgfx

Configuration

A connection is configured from the environment, under the prefix the application gives it:

POSTGRES_HOST=localhost:5432
POSTGRES_DATABASE=orders
POSTGRES_USER=orders
POSTGRES_PASSWORD=...
POSTGRES_APP_NAME=orders-api

POSTGRES_TLS_MODE=verify-full
POSTGRES_TLS_ROOT_CERT=/run/secrets/ca.crt

POSTGRES_POOL_MAX_CONNS=20
POSTGRES_TIMEOUTS_CONNECT=5s

POSTGRES_HOST and POSTGRES_DATABASE must be supplied. When USER or PASSWORD is empty, pgx uses its normal libpq-compatible defaults. Every pool and timeout field left unset keeps the pgxpool default, so a deployment sets only what it means to change.

confx.Manifest[pgfx.Config]("postgres") lists every variable a connection reads, with its type and default.

For verified TLS, use verify-full with TLS_ROOT_CERT. Use WithTLS when the application already has an in-memory *tls.Config:

pgfx.WithTLS(tlsConfig)

Fx

Load one default connection:

fx.New(
	confx.Module(),
	confx.Provide[pgfx.Config]("postgres"),
	pgfx.Module,
)

pgfx.Module provides an untagged *pgxpool.Pool, verifies the connection during startup, and closes the pool during shutdown.

Add a named replica or shard - the name gives both the Fx tag and the prefix, so replica reads REPLICA_HOST and the rest:

fx.New(
	confx.Module(),
	confx.Provide[pgfx.Config]("postgres"),
	confx.ProvideNamed[pgfx.Config]("replica"),
	pgfx.Module,
	pgfx.ModuleFor("replica"),
)

A replica inherits nothing from the default connection: its database, user and TLS settings are given again under its own prefix.

Without Fx:

pool, err := pgfx.Make(ctx, cfg)

Queries

The generic helpers accept a pool, connection, or transaction:

order, err := pgfx.FetchRow[Order](
	ctx,
	pool,
	`SELECT id, amount FROM orders WHERE id = $1`,
	id,
)

id, err := pgfx.FetchValue[int64](
	ctx,
	pool,
	`INSERT INTO orders (amount) VALUES ($1) RETURNING id`,
	amount,
)

Available helpers:

  • FetchRows[T]
  • FetchRow[T]
  • FetchValues[T]
  • FetchValue[T]

Transactions

err := pgfx.Tx(ctx, pool, pgx.TxOptions{}, func(tx pgx.Tx) error {
	_, err := pgfx.FetchValue[int64](
		ctx,
		tx,
		`INSERT INTO orders (amount) VALUES ($1) RETURNING id`,
		amount,
	)

	return err
})

A nil callback error commits the transaction; a non-nil error rolls it back.

Tracing

When a trace.TracerProvider is available in Fx, pgfx.Module enables OpenTelemetry query spans automatically.

Without Fx:

pool, err := pgfx.Make(
	ctx,
	cfg,
	pgfx.WithTracing(
		otelpgx.WithTracerProvider(provider),
		otelpgx.WithTrimSQLInSpanName(),
	),
)

A custom pgx.QueryTracer can be added with WithTracer.

Query metrics

Record query duration through a callback:

fx.Supply(pgfx.QueryMetricFunc(func(metric pgfx.QueryMetric) {
	// metric.Name, metric.Duration, metric.Err
}))

Attach a stable query name through the context:

ctx = pgfx.WithQueryName(ctx, "orders.get_by_id")

Raw SQL is excluded by default. Enable it only when safe:

pgfx.WithSQLInQueryMetrics(true)

Runtime options

Additional options include:

  • WithBeforeConnect
  • WithAfterConnect
  • WithDefaultQueryExecMode
  • WithStatementCacheCapacity
  • WithRuntimeParam

Acknowledgements

I am grateful to the authors of pgx, Uber Fx, and otelpgx. Their work made this library possible.

License

MIT

Documentation

Overview

Package pgfx is a thin, Fx-first layer over pgx/pgxpool for Postgres. A pgfx connection is a single *pgxpool.Pool built from a plain, serializable Config. Runtime dependencies (tracer, query metrics, pool hooks, an in-memory *tls.Config) are supplied through Option values, never through Config.

pgfx is meant to be wired with Uber Fx. Module is the default single connection and ModuleFor adds a replica or another shard; both consume a pgfx.Config the application supplies (typically via confmaker/confx) and manage the pool's lifecycle. pgfx does not read any config source itself, so it stays decoupled from confmaker. Multiple databases are simply multiple named connections. To build a pool without Fx, call Make with a Config you fill yourself.

Index

Constants

This section is empty.

Variables

View Source 
var Module = fx.Module("pgfx", poolProvider(``))

Module is an Fx module for the default Postgres connection: it consumes an untagged pgfx.Config from the container and provides an untagged *pgxpool.Pool, so a repository just depends on *pgxpool.Pool - no tags, no wrappers.

The application supplies the Config explicitly - pgfx does not read any config source itself - typically with confmaker/confx: 

fx.New(
	confx.Module(),
	confx.Provide[pgfx.Config]("postgres"),
	pgfx.Module,
)

The pool is pinged on start (fail-fast on a dead database) and closed on stop, and takes an optional trace.TracerProvider and QueryMetricFunc from the container.

Functions

func FetchRow 

func FetchRow[T any](
	ctx context.Context,
	q Querier,
	sql string,
	args ...any,
) (T, error)

FetchRow runs the query on q and decodes exactly one struct row into T. It returns sql.ErrNoRows when the query yields no rows, and pgx.ErrTooManyRows when it yields more than one. For a scalar result use FetchValue.

func FetchRows 

func FetchRows[T any](
	ctx context.Context,
	q Querier,
	sql string,
	args ...any,
) ([]T, error)

FetchRows runs the query on q and decodes every row into a struct T. Columns map to struct fields by the `db:"..."` tag (falling back to the field name). For a single-column (scalar) result use FetchValues.

func FetchValue 

func FetchValue[T any](
	ctx context.Context,
	q Querier,
	sql string,
	args ...any,
) (T, error)

FetchValue runs the query on q and decodes exactly one single-column row into a scalar T. It is the helper for count(*), INSERT ... RETURNING id and similar. It returns sql.ErrNoRows when the query yields no rows, and pgx.ErrTooManyRows when it yields more than one. For struct rows use FetchRow.

func FetchValues 

func FetchValues[T any](
	ctx context.Context,
	q Querier,
	sql string,
	args ...any,
) ([]T, error)

FetchValues runs the query on q and decodes a single-column result into a slice of scalar T (int64, string, uuid.UUID, ...). For struct rows use FetchRows.

func Make 

func Make(ctx context.Context, cfg Config, opts ...Option) (*pgxpool.Pool, error)

Make validates cfg and opens a connection pool for it, applying the runtime opts. The caller owns the returned pool and must Close it (the fx module does this via lifecycle).

func ModuleFor 

func ModuleFor(name string) fx.Option

ModuleFor is an Fx module for a named connection - a replica or another shard. It consumes a pgfx.Config tagged name:"<name>" and provides a *pgxpool.Pool tagged the same; a consumer selects it with fx.ParamTags. The application supplies the tagged Config explicitly (typically confmaker/confx's Provide).

func QueryName 

func QueryName(ctx context.Context) string

QueryName returns the name set by WithQueryName, or "" if none.

func Tx 

func Tx(
	ctx context.Context,
	db TxBeginner,
	opts pgx.TxOptions,
	fn func(pgx.Tx) error,
) (err error)

Tx runs fn inside a transaction begun on db and commits it, or rolls back on error or panic. The transaction is passed explicitly to fn (no ambient transaction via context): because pgx.Tx satisfies Querier, FetchRow/FetchRows work on it unchanged.

opts selects the isolation level (e.g. pgx.TxOptions{IsoLevel: pgx.Serializable}).

Tx does not retry serialization failures (SQLSTATE 40001) and does not nest: calling Tx from within fn opens a second top-level transaction on a different connection, not a savepoint. Handle both concerns at the call site if needed.

func WithQueryName 

func WithQueryName(ctx context.Context, name string) context.Context

WithQueryName tags ctx with a bounded query name used as QueryMetric.Name. Set it per call site (e.g. "orders.get_by_id"); keep the set of names small.

Types

type Config 

type Config struct {
	// Host is the endpoint as "host" or "host:port". A port in the string wins;
	// when omitted, 5432 applies. IPv6 with a port must be bracketed
	// ("[::1]:5433").
	Host string `env:"HOST,notEmpty"`
	// Database is the PostgreSQL database name.
	Database string `env:"DATABASE,notEmpty"`
	// User is the PostgreSQL role. When empty, libpq selects its default.
	User string `env:"USER"`
	// Password is read from the environment only. When empty, libpq may use
	// PGPASSWORD or .pgpass.
	Password secret.Secret `env:"PASSWORD"`
	// AppName is reported as application_name in pg_stat_activity.
	AppName string `env:"APP_NAME"`

	// TLS configures libpq transport security.
	TLS TLSConfig `envPrefix:"TLS_"`
	// Pool configures pgxpool sizing and connection lifetimes.
	Pool PoolConfig `envPrefix:"POOL_"`
	// Timeouts configures connection timeouts.
	Timeouts TimeoutConfig `envPrefix:"TIMEOUTS_"`
}

Config is the serializable configuration of a single Postgres connection. It carries only plain, serializable values - no tracer, hooks, or other runtime dependencies (those are passed to Make as Option values).

The `env` tags let a loader fill it from the environment (see confmaker/confx). They are inert strings, so the Config type itself depends only on the standard-library-only github.com/uchaloop/secret/v2 module.

Host and Database must be supplied by the deployment. User and Password are optional: when empty they fall back to libpq's defaults (PGUSER / the OS user, and PGPASSWORD / .pgpass).

func (Config) Validate 

func (cfg Config) Validate() error

Validate checks what the values mean, once a loader has supplied them: every problem at once, not just the first, so a misconfigured deployment takes one rollout to fix rather than one per mistake.

type Option 

type Option func(*options)

Option configures the runtime behaviour of a connection built by Make.

func WithAfterConnect 

func WithAfterConnect(fn func(context.Context, *pgx.Conn) error) Option

WithAfterConnect installs pgxpool.Config.AfterConnect - the usual place for custom pgx type registration.

func WithBeforeConnect 

func WithBeforeConnect(fn func(context.Context, *pgx.ConnConfig) error) Option

WithBeforeConnect installs pgxpool.Config.BeforeConnect - invoked before establishing each new connection (e.g. to rotate a password).

func WithDefaultQueryExecMode 

func WithDefaultQueryExecMode(mode pgx.QueryExecMode) Option

WithDefaultQueryExecMode sets the default pgx.QueryExecMode.

func WithQueryMetrics 

func WithQueryMetrics(fn QueryMetricFunc) Option

WithQueryMetrics installs a second pgx.QueryTracer that reports a QueryMetric for every query (composed with the span tracer). nil disables it.

func WithRuntimeParam 

func WithRuntimeParam(key, value string) Option

WithRuntimeParam adds a single pgx.ConnConfig.RuntimeParams value (SET key = value on connect).

func WithSQLInQueryMetrics 

func WithSQLInQueryMetrics(include bool) Option

WithSQLInQueryMetrics includes the raw SQL in QueryMetric.SQL (default off). Enable only when statements are safe to expose.

func WithStatementCacheCapacity 

func WithStatementCacheCapacity(capacity int) Option

WithStatementCacheCapacity sets the statement-cache size per connection.

func WithTLS 

func WithTLS(tlsConfig *tls.Config) Option

WithTLS sets the connection's *tls.Config directly, overriding whatever the file-based Config.TLS produced (sslmode and the cert/key/root-cert paths). Use it when certificates are not files on disk - loaded from a secret manager, held in memory, or rotated at runtime: the application builds the *tls.Config and passes it here. It also removes libpq TLS/plaintext fallbacks so every connection attempt uses exactly this configuration. A nil value disables TLS.

func WithTracer 

func WithTracer(tracer pgx.QueryTracer) Option

WithTracer adds a custom pgx.QueryTracer (for example an AWS X-Ray tracer). It composes with WithTracing and WithQueryMetrics rather than replacing them.

func WithTracing 

func WithTracing(opts ...otelpgx.Option) Option

WithTracing enables OpenTelemetry query spans via otelpgx. Tracing is off by default; this turns it on. Configure it with otelpgx's own options, e.g. WithTracing(otelpgx.WithTracerProvider(tp), otelpgx.WithTrimSQLInSpanName()).

type PoolConfig 

type PoolConfig struct {
	// MaxConns is the maximum pool size.
	MaxConns int32 `env:"MAX_CONNS"`
	// MinConns is the minimum number of connections maintained by the pool.
	MinConns int32 `env:"MIN_CONNS"`
	// MinIdleConns is the minimum number of idle connections maintained by the
	// pool.
	MinIdleConns int32 `env:"MIN_IDLE_CONNS"`
	// MaxConnLifetime is the maximum lifetime of a connection.
	MaxConnLifetime time.Duration `env:"MAX_CONN_LIFETIME"`
	// MaxConnLifetimeJitter randomizes connection expiry to avoid synchronized
	// reconnects.
	MaxConnLifetimeJitter time.Duration `env:"MAX_CONN_LIFETIME_JITTER"`
	// MaxConnIdleTime is the maximum time a connection may remain idle.
	MaxConnIdleTime time.Duration `env:"MAX_CONN_IDLE_TIME"`
	// HealthPeriod controls how often pgxpool checks idle connections.
	HealthPeriod time.Duration `env:"HEALTH_PERIOD"`
}

PoolConfig holds pgxpool sizing and connection-lifecycle settings. Zero values leave the pgxpool defaults in place.

type Querier 

type Querier interface {
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
}

Querier is the minimal query surface the generic fetch helpers need. It is satisfied by *pgxpool.Pool, *pgx.Conn and pgx.Tx alike, so the helpers run unchanged both directly on a pool and inside a transaction.

type QueryMetric 

type QueryMetric struct {
	// Name is the bounded label from WithQueryName ("" if unset) - use it as the
	// metric label, never SQL (raw statements have unbounded cardinality).
	Name string
	// SQL is the raw statement when explicitly enabled with WithSQLInQueryMetrics;
	// otherwise it is empty. Do NOT use it as a metric label, and scrub it before
	// recording.
	SQL string
	// Duration is the wall time from query start to end.
	Duration time.Duration
	// RowsAffected is the number of rows affected or returned as reported by pgx.
	RowsAffected int64
	// Err is non-nil when the query failed.
	Err error
}

QueryMetric is one observed query, passed to a QueryMetricFunc.

type QueryMetricFunc 

type QueryMetricFunc func(QueryMetric)

QueryMetricFunc receives one QueryMetric per query. It runs on the query hot path, so it must be cheap and non-blocking. The package ships no metrics dependency - this callback is the seam where you plug one in.

type TLSConfig 

type TLSConfig struct {
	// Mode is the libpq sslmode (disable/allow/prefer/require/verify-ca/
	// verify-full). Empty leaves the libpq default ("prefer").
	//
	// SECURITY: "prefer"/"allow" fall back to unencrypted and never verify the
	// server certificate; "require" encrypts but still does not verify it. Use
	// "verify-full" with RootCert for MITM protection.
	Mode string `env:"MODE"`
	// Cert is the path to the client certificate.
	Cert string `env:"CERT"`
	// Key is the path to the client private key.
	Key string `env:"KEY"`
	// RootCert is the path to the trusted root certificate.
	RootCert string `env:"ROOT_CERT"`
	// ServerName overrides the TLS server name used for verification.
	ServerName string `env:"SERVER_NAME"`
}

TLSConfig holds the libpq TLS settings.

type TimeoutConfig 

type TimeoutConfig struct {
	// Connect bounds establishing a single connection. Zero leaves the pgx
	// default.
	Connect time.Duration `env:"CONNECT"`
}

TimeoutConfig holds connection timeouts.

type TxBeginner 

type TxBeginner interface {
	BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
}

TxBeginner starts transactions. It is satisfied by *pgxpool.Pool and *pgx.Conn, so Tx runs on a pool or a single connection alike - the transaction-level counterpart of Querier for the fetch helpers.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.