Help us shape the future of Go! Take the 2026 Go Developer Survey

database

package
v1.2.0 Latest Latest
Warning

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

 Go to latest Published: Sep 30, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

Documentation ¶

Overview ¶

Package database provides PostgreSQL-first database infrastructure for Hellnet Go services. Configuration is environment-first, modular and cloud-native. 

env vars → Options → pgxpool.Pool → *DB / Repository[T]

For conventional database/sql access, use NewSQLX. The pgx-backed DB stays available for native COPY, batching and LISTEN/NOTIFY operations.

The package mirrors the Hellnet .NET library Hellnet.Database:

  • Executor-style raw SQL (Execute, Query[T], QueryRow[T], Scalar[T])
  • Automatic transactions with commit/rollback (Transactional)
  • A generic Repository[T] with specification-based queries
  • Typed results (Result[T], PageResult[T])
  • Transient-failure retry with SQLSTATE discrimination

There is no ORM: structs are mapped by name using `db:"column"` tags.

Index ¶

Examples ¶

Constants ¶

View Source 
const (
	OpQuery    = "query"     // Query[T] / TxQuery[T] / ConnQuery[T]
	OpQueryRow = "query_row" // QueryRow[T] / TxQueryRow[T] / ConnQueryRow[T]
	OpScalar   = "scalar"    // Scalar[T] / TxScalar[T] / ConnScalar[T]
	OpExec     = "exec"      // Execute
)

Op values usados em QueryInfo.Op (e nos labels `op` das métricas).

Variables ¶

View Source 
var ErrInvalidView = errors.New("database: DBView not initialized; use DB.WithCommandTimeout")

ErrInvalidView is returned by every DBView operation invoked on a nil or zero-value view (i.e. not produced via WithCommandTimeout).

View Source 
var ErrNilPrimary = errors.New("database: cluster requires a non-nil primary DB")

ErrNilPrimary is returned by NewCluster when the primary is missing.

View Source 
var ErrNoRows = pgx.ErrNoRows

ErrNoRows is re-exported so callers do not need to import pgx directly for the most common sentinel check.

Functions ¶

func ClusterQuery ¶ 

func ClusterQuery[T any](cl *Cluster, sql string, args ...any) ([]T, error)

ClusterQuery runs a SELECT against the next round-robin REPLICA (or the primary when none are configured) mapping rows into T.

func ClusterQueryRow ¶ 

func ClusterQueryRow[T any](cl *Cluster, sql string, args ...any) (T, bool, error)

ClusterQueryRow runs at-most-one-row reads on the next REPLICA; empty results yield (zero, false, nil), matching QueryRow[T].

func ClusterScalar ¶ 

func ClusterScalar[T any](cl *Cluster, sql string, args ...any) (T, error)

ClusterScalar scans a single value from the next REPLICA.

func ConnIter ¶ 

func ConnIter[T any](c *Conn, sql string, args []any, fn func(row T) error) error

ConnIter is Iter pinned to a dedicated connection (no retry either).

func ConnQuery ¶ 

func ConnQuery[T any](c *Conn, sql string, args ...any) ([]T, error)

ConnQuery maps every row into T on the dedicated connection. No retry.

func ConnQueryContext ¶ added in v1.1.1 

func ConnQueryContext[T any](ctx context.Context, c *Conn, sql string, args ...any) ([]T, error)

ConnQueryContext maps rows using the caller's context. No retry is performed.

func ConnQueryRow ¶ 

func ConnQueryRow[T any](c *Conn, sql string, args ...any) (T, bool, error)

ConnQueryRow runs a query expected to return at most one row on the dedicated connection. No retry.

func ConnQueryRowContext ¶ added in v1.1.1 

func ConnQueryRowContext[T any](ctx context.Context, c *Conn, sql string, args ...any) (T, bool, error)

ConnQueryRowContext maps one row using the caller's context. No retry is performed.

func ConnScalar ¶ 

func ConnScalar[T any](c *Conn, sql string, args ...any) (T, error)

ConnScalar scans a single value on the dedicated connection. No retry.

func ConnScalarContext ¶ added in v1.1.1 

func ConnScalarContext[T any](ctx context.Context, c *Conn, sql string, args ...any) (T, error)

ConnScalarContext scans one value using the caller's context. No retry is performed.

func DoValue ¶ 

func DoValue[T any](p RetryPolicy, fn func() (T, error)) (T, error)

DoValue é o irmão com valor de Do: executa fn retornando (T, error) sob a mesma política — backoff exponential idêntico, discriminação de SQLSTATEs permanentes idêntica e execução única quando desabilitada. Exposto no nível de pacote para que OUTRAS libs/apps reutilizem a política sem passar pela camada de conexão: 

got, err := database.DoValue(policy, func() (string, error) { ... })

func EnsureTable ¶ 

func EnsureTable(db *DB, name, ddl string) error

EnsureTable executes an idempotent CREATE TABLE statement supplied in FULL by the caller — ddl must include the leading CREATE TABLE and carry IF NOT EXISTS so repeated service startups stay no-ops. The table name is validated as a plain identifier first, mirroring NewRepositoryForTable's injection guard; when the DDL does not spell out the same name the call is still safe (validation is advisory), but passing mismatched inputs is a programming slip worth failing loudly on.

func Iter ¶ 

func Iter[T any](db *DB, sql string, args []any, fn func(row T) error) error

Iter streams rows of sql through fn with constant memory usage, on the pooled DB. Unlike the typed queries, streams are NEVER retried: a partially consumed stream cannot be replayed inside fn without duplicating side effects. See runIter for timeout/abort semantics. 

SQL string, args []any positional slice — pass nil when there are none.

func Query ¶ 

func Query[T any](db *DB, sql string, args ...any) ([]T, error)

Query runs a SELECT and maps every row into a T. Transient failures are retried when retry is enabled. Every retry attempt fires QueryHooks again. Emits a db.query OTel span when telemetry is attached.

Example ¶

ExampleQuery runs raw SQL mapped into typed rows. 

var _ = func(db *DB) error { //nolint:staticcheck // documentation example
	pending, err := Query[Order](db,
		"SELECT * FROM orders WHERE status = $1", "pending")
	if err != nil {
		return err
	}
	fmt.Println(len(pending))
	return nil
}
return

func QueryContext ¶ added in v1.1.1 

func QueryContext[T any](ctx context.Context, db *DB, sql string, args ...any) (out []T, err error)

QueryContext runs a typed query as a child of ctx.

func QueryRow ¶ 

func QueryRow[T any](db *DB, sql string, args ...any) (T, bool, error)

QueryRow runs a query expected to return at most one row. Transient failures are retried when retry is enabled. Every retry attempt fires QueryHooks again. Emits a db.query OTel span when telemetry is attached.

func QueryRowContext ¶ added in v1.1.1 

func QueryRowContext[T any](ctx context.Context, db *DB, sql string, args ...any) (out T, found bool, err error)

QueryRowContext runs a single-row query as a child of ctx.

func Scalar ¶ 

func Scalar[T any](db *DB, sql string, args ...any) (T, error)

Scalar runs a single-value query and scans it into T. Transient failures are retried when retry is enabled. Every retry attempt fires QueryHooks again.

func ScalarContext ¶ added in v1.1.1 

func ScalarContext[T any](ctx context.Context, db *DB, sql string, args ...any) (out T, err error)

ScalarContext runs a scalar query as a child of ctx.

func TxIter ¶ 

func TxIter[T any](tx *Tx, sql string, args []any, fn func(row T) error) error

TxIter is Iter inside a transaction: rows see the tx's uncommitted state.

func TxQuery ¶ 

func TxQuery[T any](tx *Tx, sql string, args ...any) ([]T, error)

TxQuery maps every row into T. Rows see the transaction's uncommitted state.

func TxQueryContext ¶ added in v1.1.1 

func TxQueryContext[T any](ctx context.Context, tx *Tx, sql string, args ...any) ([]T, error)

TxQueryContext maps rows inside the transaction using the caller's context.

func TxQueryRow ¶ 

func TxQueryRow[T any](tx *Tx, sql string, args ...any) (T, bool, error)

TxQueryRow runs a query expected to return at most one row inside the transaction.

func TxQueryRowContext ¶ added in v1.1.1 

func TxQueryRowContext[T any](ctx context.Context, tx *Tx, sql string, args ...any) (T, bool, error)

TxQueryRowContext maps one row inside the transaction using the caller's context.

func TxScalar ¶ 

func TxScalar[T any](tx *Tx, sql string, args ...any) (T, error)

TxScalar scans a single-value result inside the transaction.

func TxScalarContext ¶ added in v1.1.1 

func TxScalarContext[T any](ctx context.Context, tx *Tx, sql string, args ...any) (T, error)

TxScalarContext scans one value inside the transaction using the caller's context.

func Validate ¶ 

func Validate(o Options) error

Validate reports every missing or invalid option as a single error, mirroring the .NET DatabaseEnvBinder.Validate behavior.

func ViewQuery ¶ 

func ViewQuery[T any](v *DBView, sql string, args ...any) ([]T, error)

ViewQuery runs a SELECT mapping every row into T under the view's timeout. Package-level (not a method) because Go methods cannot declare type parameters — the same constraint that shaped Query[T]/TxQuery[T].

func ViewQueryRow ¶ 

func ViewQueryRow[T any](v *DBView, sql string, args ...any) (T, bool, error)

ViewQueryRow runs at-most-one-row queries under the view's timeout; an empty result is (zero, false, nil), exactly like QueryRow[T].

func ViewScalar ¶ 

func ViewScalar[T any](v *DBView, sql string, args ...any) (T, error)

ViewScalar scans a single value under the view's timeout.

Types ¶

type Batch ¶ 

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

Batch is a queue of statements sent to PostgreSQL in ONE round trip (pipelining). Create it with Tx.Batch, append work with Queue and dispatch with Tx.SendBatch — results are then consumed strictly in queue order via BatchResults. 

b := tx.Batch()
b.Queue("INSERT ...", ...)
b.Queue("SELECT COUNT(*) ...")
br := tx.SendBatch(b)
n, err := br.ExecResult()
var count int64
err = br.RowScan(&count)
err = br.Close() // REQUIRED — see BatchResults.Close

func (*Batch) Len ¶ 

func (b *Batch) Len() int

Len reports how many statements are queued so far.

func (*Batch) Queue ¶ 

func (b *Batch) Queue(sql string, args ...any)

Queue appends one statement (with positional arguments) to the batch. It never touches the network — everything is flushed at SendBatch time.

type BatchResults ¶ 

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

BatchResults reads the outcomes of a sent batch, in queue order. Every method consumes exactly one result slot.

func (*BatchResults) Close ¶ 

func (br *BatchResults) Close() error

Close finishes the batch operation: unread results are drained, the derived context is released and the connection becomes usable again. It MUST be called after consuming results (defer it right after SendBatch) — skipping it leaves the connection stuck on the batch protocol. Safe to call more than once; subsequent calls return the first error.

func (*BatchResults) ExecResult ¶ 

func (br *BatchResults) ExecResult() (int64, error)

ExecResult consumes the next result as a command result, reporting affected rows. Errors propagate raw so SQLSTATE discrimination still works at the call site.

func (*BatchResults) RowScan ¶ 

func (br *BatchResults) RowScan(dest ...any) error

RowScan consumes the next result as a single row, scanning its columns into dest in order. ErrNoRows propagates unwrapped for errors.Is checks.

type Cluster ¶ 

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

Cluster routes writes to primary and reads across replicas.

func NewCluster ¶ 

func NewCluster(primary *DB, replicas ...*DB) (*Cluster, error)

NewCluster assembles a read-routing cluster around the mandatory primary; replicas may be empty. Every pointer is validated up front so later nil dereferences become impossible by construction (fail-fast beats panic).

func (*Cluster) Execute ¶ 

func (cl *Cluster) Execute(sql string, args ...any) (int64, error)

Execute routes to PRIMARY and returns affected rows (passthrough).

func (*Cluster) Primary ¶ 

func (cl *Cluster) Primary() *DB

Primary returns the write-owning DB.

func (*Cluster) Replica ¶ 

func (cl *Cluster) Replica() *DB

Replica returns the next read target: round-robin across configured replicas, or the primary itself when none exist. Safe under concurrency — the rotation rides an atomic counter, and multiple in-flight readers may observe overlapping indices without tearing anything.

func (*Cluster) Transactional ¶ 

func (cl *Cluster) Transactional(fn func(tx *Tx) error) error

Transactional routes to PRIMARY (passthrough): transactions always own the writable node, regardless of isolation level.

func (*Cluster) TransactionalLevel ¶ 

func (cl *Cluster) TransactionalLevel(level string, fn func(tx *Tx) error) error

TransactionalLevel routes to PRIMARY (passthrough), validating the level before any begin. See (*DB).TransactionalLevel for the full contract.

type Conn ¶ 

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

Conn is a dedicated database connection. Unlike the pooled DB, a Conn pins a single underlying connection — either borrowed from the pool via Acquire, or opened directly (no pool) via Connect — so every operation runs on the same session. This is what you want when a batch of N interactions must share one connection, when you need explicit open/close lifecycle control, or when you want to drive multiple connections yourself.

A Conn reuses the exact same typed core as DB and Tx: it has Execute (via embedding), ConnQuery[T], ConnQueryRow[T] and ConnScalar[T]. Like Tx, it NEVER retries — re-running partially applied work on a pinned connection would be incorrect, so the caller owns the retry decision for the unit.

Lifecycle: every Acquire/Connect must be paired with Close (releasing the connection back to the pool, or closing the physical link). For a transactional scope call Begin and then Commit/Rollback; keep the Conn open until the transaction is finished, then Close — exactly like raw pgx.

func Connect ¶ 

func Connect(opts Options) (*Conn, error)

Connect opens a single standalone connection (no pool) from explicit options. Use it for short-lived programs, scripts, or whenever you explicitly want exactly one connection with full open/close control. Pair with Close. Internal operations derive their timeouts from a Background context captured once here and propagated internally to every later operation.

func ConnectFromEnv ¶ 

func ConnectFromEnv() (*Conn, error)

ConnectFromEnv loads options from DATABASE_* variables (and a .env file via LoadFromEnv) and opens a single standalone connection. The env loading is fully contained in the library.

func (*Conn) Begin ¶ 

func (c *Conn) Begin() (*Tx, error)

Begin starts a transaction on this dedicated connection. The returned *Tx is committed or rolled back by the caller via Tx.Commit / Tx.Rollback. The Conn must remain open until the transaction is finished; call Close afterwards. The context captured once at construction is used internally, bounded by the configured ConnectionTimeout.

func (*Conn) BulkInsert ¶ 

func (c *Conn) BulkInsert(table string, columns []string, rows [][]any) (int64, error)

BulkInsert loads every row into table using PostgreSQL's COPY ... FROM STDIN protocol — orders of magnitude faster than INSERT loops for large volumes. The table and each column name are validated as plain identifiers first; rows must be positional and match the columns order (exactly like raw pgx CopyFrom). Returns the number of rows copied. The context captured once at New/Connect is used internally, bounded by CommandTimeout.

Only Conn exposes this by design: COPY is pinned to one connection, which is exactly what a dedicated Conn guarantees.

func (*Conn) Close ¶ 

func (c *Conn) Close() error

Close releases or closes the underlying connection. For an Acquired connection it returns to the pool; for a Connect-ed connection it closes the physical link. Safe to call more than once; only the first call has effect. The context captured once at New/Connect is used internally.

func (*Conn) Execute ¶ 

func (c *Conn) Execute(sql string, args ...any) (int64, error)

Execute runs a command on the dedicated connection and emits db.execute.

func (*Conn) ExecuteContext ¶ added in v1.1.1 

func (c *Conn) ExecuteContext(ctx context.Context, sql string, args ...any) (int64, error)

ExecuteContext executes a command on the dedicated connection as a child of ctx.

func (*Conn) Listen ¶ 

func (c *Conn) Listen(channel string, handler func(payload string)) (stop func() error, err error)

Listen registers handler for async notifications published on channel and returns immediately after the initial LISTEN succeeds. On the first fatal connection error the fetch loop gives up (the Warn is logged); use ListenWithReconnect when the delivery must survive dropped connections.

The returned stop releases everything: it ends the fetch goroutine (~1s at most, one notifyTick) and issues a best-effort UNLISTEN on the session. Handler invocations run on their own goroutines so a slow handler never blocks deliveries; note stop() does NOT wait for an in-flight handler.

The context captured once at New/Connect is used internally; each fetch iteration is bounded by a 1s tick instead of CommandTimeout, since waiting for events is expected to outlive any command deadline.

func (*Conn) ListenWithReconnect ¶ 

func (c *Conn) ListenWithReconnect(channel string, handler func(payload string), opts ListenOptions) (stop func() error, err error)

ListenWithReconnect is Listen with a re-listen policy: when the fetch loop hits a connection error it re-executes `LISTEN channel` forever, with exponential backoff from opts.RetryDelay up to opts.MaxBackoff, logging each attempt at WARN level, until stop() is called. Note this re-listens the SAME pinned connection — if the physical link died permanently the attempts keep failing until the caller stops and rebuilds the Conn; it covers transient interruptions where pgx can keep using the session.

func (*Conn) Notify ¶ 

func (c *Conn) Notify(channel string, payload string) error

Notify publishes payload on channel through `SELECT pg_notify($1,$2)` (bind parameters ARE supported here). The channel is validated like Listen. PostgreSQL limits payloads to 8000 bytes — larger ones fail server-side. The context captured once at New/Connect is used internally, bounded by CommandTimeout.

func (*Conn) Transactional ¶ 

func (c *Conn) Transactional(fn func(tx *Tx) error) error

Transactional runs fn atomically on this dedicated connection: fn's operations share the connection and commit together, rolling back on error. The context captured once at New/Connect is propagated internally to begin, fn's statements, commit and rollback. The Conn itself stays open after the call — call Close when finished with it.

func (*Conn) TransactionalContext ¶ added in v1.1.1 

func (c *Conn) TransactionalContext(ctx context.Context, fn func(ctx context.Context, tx *Tx) error) error

TransactionalContext runs an atomic transaction as a child of ctx.

func (*Conn) TransactionalLevel ¶ 

func (c *Conn) TransactionalLevel(level string, fn func(*Tx) error) error

TransactionalLevel is Transactional with an explicit isolation level ("read committed", "repeatable read" or "serializable"). Both *pgx.Conn and *pgxpool.Conn satisfy BeginTx-with-options natively (asserted here), so no connection-type branching leaks into callers. Invalid levels fail fast before any begin is attempted; the Conn stays open after the call exactly like plain Transactional.

type ContextQueryHook ¶ added in v1.1.1 

type ContextQueryHook interface {
	BeforeQueryContext(context.Context, QueryInfo)
	AfterQueryContext(context.Context, QueryInfo)
}

ContextQueryHook is an optional extension to QueryHook. Implementations can correlate hook work with the request trace without changing QueryHook's public contract.

type DB ¶ 

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

DB is the entry point of the library: a pooled PostgreSQL connection with executor methods, transactional support and retry semantics. Create it with New, MustNew or OpenFromEnv; it is safe for concurrent use.

DB embeds conn (runner + options) so the typed core is shared with Tx; the pool field is kept for lifecycle operations (Begin/Ping/Close) outside the runner surface.

func MustNew ¶ 

func MustNew(ctx context.Context, inst instrument.Instrumentation, opts ...Options) *DB

MustNew is like New but panics on failure. Useful at service startup where a misconfiguration should fail fast.

func New ¶ 

func New(ctx context.Context, inst instrument.Instrumentation, opts ...Options) (*DB, error)

New creates a DB. With no options it loads DATABASE_* (and a .env file) through LoadFromEnv; pass a single Options to configure it explicitly. inst is the Hellnet observability contract (for example a *telemetry.Telemetry); a nil inst disables telemetry. The context is captured once and used for the initial connection setup and for the context-less compatibility methods. No connection is established yet; call PingContext to verify.

func NewWithOptions ¶ added in v1.0.5 

func NewWithOptions(ctx context.Context, opts ...Options) (*DB, error)

NewWithOptions accepts legacy Options values and the additive WithInstrumentation option.

func OpenFromEnv deprecated 

func OpenFromEnv(ctx context.Context, inst instrument.Instrumentation) (*DB, error)

OpenFromEnv loads options from the environment (and a .env file) and builds the DB. It is the equivalent of the .NET AddHellnetDatabase() env-first overload and is a thin wrapper over New. The env loading is fully contained in the library — no external DotEnv call is required by the caller.

Deprecated: use New, which is env-first when called without options.

Example ¶

ExampleOpenFromEnv connects using DATABASE_* variables. 

// Context is captured once at construction and propagated internally.
db, err := OpenFromEnv(context.Background(), nil)
if err != nil {
	slog.Error("configuration error", "err", err)
	return
}
defer func() { _ = db.Close() }()

count, err := Scalar[int64](db, "SELECT COUNT(*) FROM orders")
if err != nil {
	slog.Error("query failed", "err", err)
	return
}
fmt.Println(count)

func (*DB) Acquire ¶ 

func (db *DB) Acquire() (*Conn, error)

Acquire borrows a single connection from the pool and pins it to the returned Conn until Close is called. Multiple Acquire calls return independent connections, so explicit multi-connection scenarios work alongside the pool's automatic multiplexing. The context captured once at New is used internally, bounded by the configured ConnectionTimeout.

func (*DB) Close ¶ 

func (db *DB) Close() error

Close releases the underlying pool.

func (*DB) EnableMetrics ¶ 

func (db *DB) EnableMetrics(reg prometheus.Registerer) (*MetricsHandle, error)

EnableMetrics registra o MetricsCollector nativo (Prometheus puro, sem OTel) como QueryHook deste DB — valendo para todas as transações/conexões criadas depois — inicia o sampler dos gauges de pool e devolve um handle cujo Close() interrompe tudo de forma idempotente.

Regras:

  • Segunda chamada num mesmo DB retorna erro ("already enabled"); após handle.Close() é permitido habilitar de novo.
  • reg nil usa prometheus.DefaultRegisterer (evite: prefira registro isolado para testes/exposição controlada).
  • O chamador é dono do handle: feche-o antes do app encerrar para não deixar goroutine nem métricas órfãs no registro.

func (*DB) Execute ¶ 

func (db *DB) Execute(sql string, args ...any) (int64, error)

Execute runs a command (INSERT/UPDATE/DELETE/DDL) and returns the number of affected rows. Transient failures are retried when retry is enabled. Every retry attempt fires QueryHooks again (one before/after pair per attempt). Emits a db.execute OTel span when telemetry is attached.

func (*DB) ExecuteContext ¶ added in v1.1.1 

func (db *DB) ExecuteContext(ctx context.Context, sql string, args ...any) (n int64, err error)

ExecuteContext executes a command as a child of ctx.

func (*DB) Options ¶ 

func (db *DB) Options() Options

Options returns the effective configuration (without the password).

func (*DB) Ping deprecated 

func (db *DB) Ping() error

Ping verifies database connectivity using the construction context.

Deprecated: use PingContext with the caller's context.

func (*DB) PingContext ¶ added in v1.1.1 

func (db *DB) PingContext(ctx context.Context) error

PingContext verifies database connectivity as a child of ctx. Register it in the service's readiness endpoint when database connectivity is required.

func (*DB) Transactional ¶ 

func (db *DB) Transactional(fn func(tx *Tx) error) error

Transactional runs fn atomically on the pooled connection: fn's operations share one connection and commit together. If fn returns an error the transaction is rolled back and the error is returned; a rollback failure is joined into the returned error. The context captured once at New is propagated internally (CommandTimeout bounds the terminal statements, ConnectionTimeout bounds Begin). This mirrors the .NET IDatabaseTransaction.ExecuteAsync contract.

Example ¶

ExampleDB_Transactional moves money between accounts atomically; any error rolls the whole unit of work back. 

var _ = func(db *DB) error { //nolint:staticcheck // documentation example
	return db.Transactional(func(tx *Tx) error {
		if _, err := tx.Execute(
			"UPDATE accounts SET balance = balance - 100 WHERE id = $1", 1); err != nil {
			return err
		}
		if _, err := tx.Execute(
			"UPDATE accounts SET balance = balance + 100 WHERE id = $1", 2); err != nil {
			return err
		}
		total, err := TxScalar[int64](tx, "SELECT SUM(balance) FROM accounts")
		if err != nil {
			return err
		}
		fmt.Println(total)
		return nil // commit
	})
}
return

func (*DB) TransactionalContext ¶ added in v1.1.1 

func (db *DB) TransactionalContext(ctx context.Context, fn func(ctx context.Context, tx *Tx) error) error

TransactionalContext runs an atomic transaction as a child of ctx.

func (*DB) TransactionalLevel ¶ 

func (db *DB) TransactionalLevel(level string, fn func(*Tx) error) error

TransactionalLevel runs fn atomically at an explicit isolation level ("read committed", "repeatable read" or "serializable"; validated first). It mirrors DB.Transactional end-to-end — shared compensation logic, retry excluded inside transactions, contexts captured once at New — while plumb- ing the level through pool.BeginTx so the requested PG semantics reach the server untouched. Invalid levels fail fast BEFORE any connection work.

func (*DB) WithCommandTimeout ¶ 

func (db *DB) WithCommandTimeout(d time.Duration) *DBView

WithCommandTimeout returns a view of db whose every statement runs under d instead of the pool-wide CommandTimeout. Values below minCommandTimeout (including zero/negative) are clamped to the floor with a single warning: the signature cannot fail, and silent acceptance would turn a config typo into either instant deadlines or unbounded waits.

type DBView ¶ 

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

DBView aliases one *DB with an effective CommandTimeout. It carries no resources of its own: Close/Ping/transactions belong to the parent.

func (*DBView) CommandTimeout ¶ 

func (v *DBView) CommandTimeout() time.Duration

CommandTimeout reports the effective timeout used by this view's statements.

func (*DBView) Execute ¶ 

func (v *DBView) Execute(sql string, args ...any) (int64, error)

Execute runs a command under the view's timeout. Retry behavior mirrors the parent DB (same policy, same captured base context bounding backoff).

type ListenOptions ¶ 

type ListenOptions struct {
	// RetryDelay is the first backoff delay after a failed re-listen attempt.
	// <= 0 means 500ms.
	RetryDelay time.Duration
	// MaxBackoff caps the exponential growth between attempts. <= 0 means 5s.
	MaxBackoff time.Duration
}

ListenOptions tunes the re-listen policy used by ListenWithReconnect.

type MetricsCollector ¶ 

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

MetricsCollector é um QueryHook embutido que traduz o fluxo de operações em métricas Prometheus nativas. Crie com NewMetricsCollector e habilite via DB.EnableMetrics (fluxo recomendado), ou registre os collectors manualmente via Collect(). A instância implementa AMBOS os hooks.

func NewMetricsCollector ¶ 

func NewMetricsCollector(reg prometheus.Registerer) *MetricsCollector

NewMetricsCollector monta o collector pronto para uso. reg nil assume prometheus.DefaultRegisterer como alvo de registro/unregistro; passar um registro próprio (prometheus.NewRegistry) isola as métricas do global. A construção NÃO registra nada — o registro acontece em DB.EnableMetrics (ou manualmente via Collect()).

func (*MetricsCollector) AfterHook ¶ 

func (mc *MetricsCollector) AfterHook(info QueryInfo)

AfterHook materializa a observação: incrementa db_queries_total{op,status} e observa a duração em db_query_duration_seconds{op}. O status é derivado do erro (ok | canceled p/ contexto morto | error para todo o resto).

func (*MetricsCollector) BeforeHook ¶ 

func (mc *MetricsCollector) BeforeHook(QueryInfo)

BeforeHook é propositalmente no-op: a duração chega pronta no AfterHook (QueryInfo.Duration), medida pela própria lib na volta da execução.

func (*MetricsCollector) Collect ¶ 

func (mc *MetricsCollector) Collect() []prometheus.Collector

Collect devolve todos os collectors gerenciados para registro/gerência externa (útil quando o chamador prefere registrar por conta própria).

type MetricsHandle ¶ 

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

MetricsHandle controla o ciclo de vida das métricas habilitadas por EnableMetrics: Close interrompe o sampler do pool, desliga o hook e desfaz o registro no Registerer informado. É idempotente (sync.Once) — chamar mais de uma vez não tem efeito. Depois de Close o DB continua funcionando normalmente; também é possível habilitar métricas novamente.

func (*MetricsHandle) Close ¶ 

func (h *MetricsHandle) Close()

Close para o sampler, desanexa o hook e desfaz o registro das métricas. Idempotente e nil-safe.

func (*MetricsHandle) Collector ¶ 

func (h *MetricsHandle) Collector() *MetricsCollector

Collector devolve o collector subjacente (leitura avançada de métricas em testes ou exposition customizado).

type Options ¶ 

type Options struct {
	// ── Connection ──────────────────────────────────────────────
	Host     string
	Port     int
	Database string
	Username string
	Password string

	// ── Pool ────────────────────────────────────────────────────
	PoolMinSize int
	PoolMaxSize int

	// ── Timeouts ────────────────────────────────────────────────
	CommandTimeout    time.Duration
	ConnectionTimeout time.Duration

	// ── Resilience ──────────────────────────────────────────────
	RetryEnabled   bool
	RetryMaxCount  int
	RetryBaseDelay time.Duration

	// ── Diagnostics ─────────────────────────────────────────────
	SlowQuery time.Duration

	// ── Observability ───────────────────────────────────────────
	// QueryHooks observam toda operação executada (before/after por tentativa
	// de retry). Zero-value friendly: nil desativa totalmente.
	QueryHooks []QueryHook
	// HideQueryArgs corta os args antes de qualquer hook/métrica
	// (PII/segredos não devem vazar para pipelines de observabilidade).
	HideQueryArgs bool
	// contains filtered or unexported fields
}

Options configures the database connection. All fields are populated from environment variables (DATABASE_*) via LoadFromEnv, or set explicitly.

func Default ¶ added in v1.0.5 

func Default() Options

Default returns the default configuration.

func LoadFromEnv ¶ 

func LoadFromEnv() Options

LoadFromEnv loads DATABASE_* environment variables (plus a .env file via loadEnvFiles) into Options, starting from Default as the fallback for any unset value. It is fully self-contained: the caller does not need to load env files beforehand.

func WithInstrumentation ¶ added in v1.1.1 

func WithInstrumentation(inst instrument.Instrumentation) Options

WithInstrumentation supplies the Hellnet observability contract. It is represented as an Options value to preserve the original variadic API.

type PageResult ¶ 

type PageResult[T any] struct {
	Items      []T
	TotalCount int64
	Page       int
	PageSize   int
}

PageResult is a paginated result. It mirrors the .NET PageResult<T>.

func (PageResult[T]) HasNextPage ¶ 

func (p PageResult[T]) HasNextPage() bool

HasNextPage reports whether there are more pages after this one.

type Pool ¶ 

type Pool interface {
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	Begin(ctx context.Context) (pgx.Tx, error)
	Ping(ctx context.Context) error
	Close()
	Stat() PoolStats
}

Pool is the minimal surface of pgxpool.Pool used by this package. It exists so unit tests can supply a fake implementation. Stat alimenta o sampler interno das métricas de pool habilitadas via DB.EnableMetrics.

type PoolStats ¶ 

type PoolStats struct {
	// InUse são conexões emprestadas (em uso na amostra).
	InUse int64
	// Idle são conexões ociosas aguardando reuso.
	Idle int64
	// Max é o limite configurado de conexões.
	Max int64
}

PoolStats é um snapshot dos contadores do pool usado pelas métricas (db_pool_in_use / db_pool_idle / db_pool_max). Desacopla o pacote do tipo concreto pgxpool.Stat sem depender de OTel.

type QueryHook ¶ 

type QueryHook interface {
	// BeforeHook é chamado antes da execução (Duration zerado).
	BeforeHook(info QueryInfo)
	// AfterHook é sempre chamado após a execução — inclusive nos caminhos de
	// erro (statement OU contexto), em par com o Before da mesma tentativa.
	// Fases são independentes: pânico contido num Before de um hook não
	// impede os outros hooks nem o After da operação.
	AfterHook(info QueryInfo)
}

QueryHook observa o ciclo de vida de cada operação. Implementações devem registrar-se via Options.QueryHooks (antes do New/Connect) ou receber o collector embutido via DB.EnableMetrics.

type QueryHookFunc ¶ 

type QueryHookFunc struct {
	Before func(QueryInfo)
	After  func(QueryInfo)
}

QueryHookFunc adapta funções anônimas ao QueryHook (zero-value friendly: campos nil simplesmente não disparam).

func (QueryHookFunc) AfterHook ¶ 

func (f QueryHookFunc) AfterHook(i QueryInfo)

AfterHook invoca f.After quando não-nil.

func (QueryHookFunc) BeforeHook ¶ 

func (f QueryHookFunc) BeforeHook(i QueryInfo)

BeforeHook invoca f.Before quando não-nil.

type QueryInfo ¶ 

type QueryInfo struct {
	// Op classifica a operação: "query"|"query_row"|"scalar"|"exec".
	Op string
	// SQL é a sentença exatamente como executada.
	SQL string
	// Args são os args posicionais; nil quando escondidos
	// (Options.HideQueryArgs) ou quando a operação não tem args.
	Args []any
	// Duration é 0 no pre-hook; no after-hook cobre statement + mapping.
	Duration time.Duration
	// Err é nil em sucesso e no pre-hook. Resultado vazio de query_row NÃO é
	// erro (semântica da lib); erro de contexto é repassado tal como veio.
	Err error
	// Rows é -1 (desconhecido) para queries sem contagem observável;
	// len(rows) em query bem-sucedida, 0/1 em query_row, rowsAffected em exec.
	Rows int64
}

QueryInfo descreve uma operação interceptada pelos hooks.

type Repository ¶ 

type Repository[T any] struct {
	// contains filtered or unexported fields
}

Repository is the generic data-access helper for one table. It mirrors the .NET IRepository<T> (plain data access, not a DDD aggregate). The table name is derived from T's type name, matching the .NET typeof(T).Name convention.

Structs should carry `db:"column"` tags so pgx maps columns correctly.

Example ¶

ExampleRepository uses the generic repository with a specification. 

var _ = func(db *DB) error { //nolint:staticcheck // documentation example
	orders := NewRepository[Order](db)

	order, found, err := orders.GetByID(42)
	if err != nil || !found {
		return err
	}
	fmt.Println(order.Status)

	spec := Specification{
		SQL:     "SELECT * FROM orders WHERE status = $1",
		Args:    []any{"pending"},
		OrderBy: "created_at DESC",
	}
	page, err := orders.Paginate(spec, 1, 20)
	if err != nil {
		return err
	}
	fmt.Println(page.TotalCount, page.HasNextPage())
	return nil
}
return

func NewRepository ¶ 

func NewRepository[T any](db *DB) *Repository[T]

NewRepository creates a repository over table name(T), mirroring the .NET typeof(T).Name convention. When the Go type name differs from the table name (the common case: Order vs orders), use NewRepositoryForTable.

func NewRepositoryForTable ¶ 

func NewRepositoryForTable[T any](db *DB, table string) *Repository[T]

NewRepositoryForTable creates a repository bound to an explicit table name, decoupling the mapping from T's type name.

func (*Repository[T]) Count ¶ 

func (r *Repository[T]) Count(spec Specification) (int64, error)

Count counts the rows matching the specification.

func (*Repository[T]) Exists ¶ 

func (r *Repository[T]) Exists(spec Specification) (bool, error)

Exists reports whether any row matches the specification, using a single EXISTS(...) round-trip instead of counting.

func (*Repository[T]) Find ¶ 

func (r *Repository[T]) Find(spec Specification) ([]T, error)

Find executes the specification's query.

func (*Repository[T]) FindOne ¶ 

func (r *Repository[T]) FindOne(spec Specification) (T, bool, error)

FindOne runs the specification expecting at most one row, appending LIMIT 1 internally (ORDER BY is honored). Returns (zero, false, nil) when nothing matches — mirroring GetByID's not-found contract.

func (*Repository[T]) GetAll ¶ 

func (r *Repository[T]) GetAll() ([]T, error)

GetAll fetches the mapped columns of every row in the table.

func (*Repository[T]) GetByID ¶ 

func (r *Repository[T]) GetByID(id any) (T, bool, error)

GetByID fetches the row with the given primary key, or (zero, false, nil) when it does not exist. The context captured once at New is used internally.

func (*Repository[T]) Paginate ¶ 

func (r *Repository[T]) Paginate(spec Specification, page, pageSize int) (PageResult[T], error)

Paginate executes the specification returning one page (1-based) plus the total row count.

func (*Repository[T]) Table ¶ 

func (r *Repository[T]) Table() string

Table returns the table name used by this repository.

func (*Repository[T]) Update ¶ 

func (r *Repository[T]) Update(entity T, setCols []string, whereSpec Specification) (int64, error)

Update assigns setCols (a REQUIRED non-empty whitelist validated against T's db-tagged mapping) from the entity's values wherever whereSpec matches. WHERE placeholders number from $1 across whereSpec.Args; SET placeholders continue after them. Returns the affected-row count. 

UPDATE tbl SET col=$N… WHERE(<whereSpec.SQL>, args…)

func (*Repository[T]) Upsert ¶ 

func (r *Repository[T]) Upsert(entity T, conflictCols, updateCols []string) (int64, error)

Upsert inserts the entity, resolving conflicts on conflictCols with an UPDATE of updateCols. Both lists are validated identifiers mapped onto T's db-tagged columns. Returns the affected-row count reported by the server: 1 on plain insert, 1 on conflict-update, 0 only when the server reports it. 

INSERT INTO tbl(cols…) VALUES($…) ON CONFLICT(conflictCols) DO UPDATE SET col=$n+…

func (*Repository[T]) UpsertDoNothing ¶ 

func (r *Repository[T]) UpsertDoNothing(entity T, conflictCols []string) (int64, error)

UpsertDoNothing is the insert-or-ignore variant: rows conflicting on conflictCols leave existing data untouched. Returns the affected-row count (0 when the row was skipped).

type Result ¶ 

type Result[T any] struct {
	Data     T
	Err      error
	Duration time.Duration
}

Result wraps a success/failure outcome with diagnostics. It mirrors the .NET DatabaseResult<T>: use Success or Failure to construct and IsSuccess to discriminate.

func Failure ¶ 

func Failure[T any](err error, duration time.Duration) Result[T]

Failure builds a failed Result.

func Success ¶ 

func Success[T any](data T, duration time.Duration) Result[T]

Success builds a successful Result.

func (Result[T]) IsSuccess ¶ 

func (r Result[T]) IsSuccess() bool

IsSuccess reports whether the operation completed without error.

type RetryPolicy ¶ 

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

RetryPolicy retries transient failures with exponential backoff. Permanent PostgreSQL errors (syntax, constraint violations, privileges, missing schema objects) are returned immediately; so is context cancellation.

func NewRetryPolicy ¶ 

func NewRetryPolicy(enabled bool, maxCount int, baseDelay time.Duration) RetryPolicy

NewRetryPolicy builds a retry policy. When enabled is false Do executes fn exactly once.

func (RetryPolicy) Do ¶ 

func (p RetryPolicy) Do(fn func() error) error

Do runs fn retrying transient failures up to maxCount times with exponential backoff. A disabled policy runs fn once. The backoff sleep is bounded by context.Background() — callers holding the library-captured base context go through do instead.

func (RetryPolicy) IsTransient ¶ 

func (p RetryPolicy) IsTransient(err error) bool

IsTransient reports whether err describes a failure worth retrying: any error except context cancellation/deadline and the permanent PostgreSQL SQLSTATEs listed in nonTransientSQLStates.

type SQLX ¶ added in v1.1.0 

type SQLX struct {
	*sqlx.DB
	// contains filtered or unexported fields
}

SQLX is the conventional SQL surface of the library. It uses sqlx with the pgx stdlib driver, so callers get database/sql compatibility and sqlx's struct mapping while the existing DB type remains available for pgx-native features such as COPY, batching and LISTEN/NOTIFY.

SQLX owns its database/sql pool and must be closed by the caller. Use DB for the existing pgx pool; the two surfaces are intentionally explicit so a service does not accidentally mix transaction types.

func NewSQLX ¶ added in v1.1.0 

func NewSQLX(ctx context.Context, opts ...Options) (*SQLX, error)

NewSQLX opens a sqlx-backed PostgreSQL pool from explicit options, or from DATABASE_* when no options are supplied. It verifies connectivity before returning, matching the fail-fast behavior of Connect.

func OpenSQLXFromEnv ¶ added in v1.1.0 

func OpenSQLXFromEnv(ctx context.Context) (*SQLX, error)

OpenSQLXFromEnv opens the sqlx pool using DATABASE_* and .env settings.

func (*SQLX) BeginTx ¶ added in v1.1.0 

func (db *SQLX) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sqlx.Tx, error)

BeginTx starts a sqlx transaction with the caller's context and options. It is declared explicitly to make the sqlx transaction boundary visible beside DB.Transactional, whose callback uses pgx.Tx.

func (*SQLX) Options ¶ added in v1.1.0 

func (db *SQLX) Options() Options

Options returns the effective SQLX configuration without the password.

type Specification ¶ 

type Specification struct {
	SQL     string
	Args    []any
	OrderBy string
}

Specification is a lean query filter: raw SQL plus positional arguments, optionally ordered. It mirrors the .NET ISpecification<T> — deliberately no expression trees.

type Tx ¶ 

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

Tx is a transaction-scoped executor. It embeds conn (carrying the live pgx.Tx as its runner and the shared options), so operations run on the same connection and inherit the common execute path. Operations on Tx never retry: re-running partially applied work would be incorrect (the caller owns the retry decision for the whole unit of work).

func (*Tx) Batch ¶ 

func (tx *Tx) Batch() *Batch

Batch creates an empty statement queue bound to this transaction. Statements execute with the transaction's snapshot: queued writes are invisible outside the tx until Commit. The context captured once at construction is used internally when the batch is dispatched.

func (*Tx) Commit ¶ 

func (tx *Tx) Commit() error

Commit commits the transaction. Use it only for transactions started via Conn.Begin, which the caller drives manually. Transactions managed by DB.Transactional / Conn.Transactional commit automatically and must NOT call this method. The context captured once at construction is used internally, bounded by CommandTimeout.

func (*Tx) Execute ¶ 

func (tx *Tx) Execute(sql string, args ...any) (int64, error)

Execute runs a command inside the transaction and emits db.execute.

func (*Tx) ExecuteContext ¶ added in v1.1.1 

func (tx *Tx) ExecuteContext(ctx context.Context, sql string, args ...any) (int64, error)

ExecuteContext executes a command inside the transaction as a child of ctx.

func (*Tx) Rollback ¶ 

func (tx *Tx) Rollback() error

Rollback aborts the transaction. Use it only for transactions started via Conn.Begin; see Commit for the ownership rules. The context captured once at construction is used internally, bounded by CommandTimeout.

func (*Tx) SendBatch ¶ 

func (tx *Tx) SendBatch(b *Batch) *BatchResults

SendBatch dispatches every queued statement in a single network round trip and returns a BatchResults to read them back in queue order. The underlying connection stays busy until BatchResults.Close is called — not calling Close leaks/holds it. SendBatch never retries (same rule as any Tx operation).

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.