Documentation
¶
Overview ¶
Package dbx provides Postgres helpers on top of jmoiron/sqlx and the pgx driver: connection setup, named query/exec wrappers with query logging, transactions, and bulk insert/upsert.
Open builds a *sqlx.DB from a Config; StatusCheck waits for it to become reachable. The Named* helpers run named queries/execs, log the expanded SQL at debug level, and translate well-known Postgres errors into the package sentinels (ErrDBNotFound, ErrDBDuplicatedEntry, ErrUndefinedTable) so callers match them with errors.Is instead of inspecting driver codes. WithinTran runs a function inside a transaction, committing on success and rolling back on error or panic. Bulk* batch large multi-row writes within Postgres's parameter limit.
Usage ¶
cfg := dbx.Config{
User: "postgres", Password: "postgres",
Host: "localhost:5432", Name: "app",
Schema: "public", MaxIdleConns: 2, MaxOpenConns: 5,
DisableTLS: true,
}
db, err := dbx.Open(cfg)
if err != nil {
return err
}
if err := dbx.StatusCheck(ctx, db); err != nil {
return err
}
// Named query into a slice.
var widgets []Widget
const q = `SELECT id, name FROM widgets WHERE name = :name`
arg := struct {
Name string `db:"name"`
}{Name: "gadget"}
if err := dbx.NamedQuerySlice(ctx, log, db, q, arg, &widgets); err != nil {
return err
}
// Pass a slice as a single array parameter: pgx binds it as a Postgres
// array, so ANY(:ids) filters without expanding placeholders.
var out []Widget
err := dbx.NamedQuerySlice(ctx, log, db,
`SELECT * FROM widgets WHERE id = ANY(:ids)`,
map[string]any{"ids": []int64{235, 401, 512}}, // slice -> one $1 -> int8[]
&out,
)
// Write inside a transaction.
err = dbx.WithinTran(ctx, log, db, func(tx *sqlx.Tx) error {
const ins = `INSERT INTO widgets (id, name) VALUES (:id, :name)`
return dbx.NamedExecContext(ctx, log, tx, ins, w)
})
if errors.Is(err, dbx.ErrDBDuplicatedEntry) {
// unique violation
}
Querying ¶
- ExecContext / NamedExecContext run statements; the RowsAffected variant returns the affected-row count.
- QueryStruct / NamedQueryStruct scan exactly one row (ErrDBNotFound when none).
- QuerySlice / NamedQuerySlice scan every row into a *[]T.
- NamedQuerySliceUsingIn / NamedQueryStructUsingIn handle a query with an IN clause: they expand a slice parameter (sqlx.In) and rebind before running.
For an IN-style filter, prefer binding the slice as a single array parameter and matching with ANY(:ids) (see Usage): pgx encodes []int64/[]string as a Postgres array, so the query text stays constant regardless of list length. Reach for the …UsingIn helpers only when you need driver-agnostic placeholder expansion instead.
The plain (non-Named) variants take no parameters; the Named variants bind from a struct via its `db` tags.
Dialects ¶
Subpackage dbx/dialect captures the few SQL fragments that differ between engines (currently pagination) behind a small Dialect interface, with Postgres and SQLite implementations, so a store can compose portable SQL.
Bulk writes ¶
BulkInsert writes len(values)/len(columns) rows, splitting into batches that stay under Postgres's bind-parameter cap, with values laid out row-major. An optional conflictAction (e.g. "ON CONFLICT DO NOTHING") is appended verbatim. BulkUpsert builds the ON CONFLICT … DO UPDATE clause for you from the conflict columns.
Transactions ¶
WithinTran is the common path. For stores that must run against either a pool or an outer transaction, depend on the Beginner / CommitRollbacker seam: NewBeginner adapts a *sqlx.DB, and ExtContext extracts the query surface from a started transaction.
Schema provisioning ¶
EnsureSchema applies idempotent DDL at startup under a transaction-scoped Postgres advisory lock, so several service replicas can provision the same table concurrently without racing — this is how the SDK storage packages (outbox, queue, auditlog, translation) create their own tables without a hand-written migration. AdvisoryKey derives a stable, collision-resistant lock key from a namespaced name (e.g. "skit.outbox.schema").
Config ¶
Config fields: User, Password, Host, Name, Schema (sets search_path), MaxIdleConns, MaxOpenConns, DisableTLS (DisableTLS=true selects sslmode=disable, otherwise sslmode=require). Open always sets timezone=utc.
Index ¶
- Constants
- Variables
- func AdvisoryKey(name string) int64
- func BulkInsert(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, table string, ...) error
- func BulkUpsert(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, table string, ...) error
- func EnsureSchema(ctx context.Context, log *logger.Logger, db *sqlx.DB, lockKey int64, ...) error
- func ExecContext(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string) error
- func ExtContext(tx CommitRollbacker) (sqlx.ExtContext, error)
- func NamedExecContext(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, ...) error
- func NamedExecContextRowsAffected(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, ...) (int64, error)
- func NamedQuerySlice[T any](ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, ...) error
- func NamedQuerySliceUsingIn[T any](ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, ...) error
- func NamedQueryStruct(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, ...) error
- func NamedQueryStructUsingIn(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, ...) error
- func Open(cfg Config) (*sqlx.DB, error)
- func OpenSQLite(cfg SQLiteConfig) (*sqlx.DB, error)
- func QuerySlice[T any](ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, ...) error
- func QueryStruct(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, ...) error
- func StatusCheck(ctx context.Context, db *sqlx.DB) error
- func WithinTran(ctx context.Context, log *logger.Logger, db *sqlx.DB, ...) error
- type Beginner
- type CommitRollbacker
- type Config
- type DBBeginner
- type JSONSlice
- type SQLiteConfig
Constants ¶
const DefaultSQLiteDriver = "sqlite"
DefaultSQLiteDriver is the driver name OpenSQLite uses when Config.Driver is empty — modernc.org/sqlite, the pure-Go driver (no cgo).
const DefaultStatusCheckTimeout = 30 * time.Second
DefaultStatusCheckTimeout bounds StatusCheck when it is called with a context that carries no deadline of its own (e.g. context.Background()). Without it a non-transient failure — wrong port, missing database, bad credentials, TLS mismatch — would retry forever and hang the caller silently. Pass a context.WithTimeout to choose a different bound.
Variables ¶
var ( ErrDBNotFound = sql.ErrNoRows ErrDBDuplicatedEntry = errors.New("duplicated entry") ErrUndefinedTable = errors.New("undefined table") )
Sentinel errors returned by this package; match them with errors.Is.
Functions ¶
func AdvisoryKey ¶ added in v0.4.0
AdvisoryKey derives a stable Postgres advisory-lock key from name using FNV-1a. It lets a package claim a collision-resistant lock key without hand-picking a magic integer that might clash with the application's own advisory locks — advisory locks share one namespace per database. Use a namespaced name such as "skit.outbox.schema".
func BulkInsert ¶
func BulkInsert(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, table string, columns []string, values []any, conflictAction string) error
BulkInsert inserts len(values)/len(columns) rows into table in batches, each batch a single multi-row INSERT. conflictAction, if non-empty, is appended verbatim (e.g. "ON CONFLICT DO NOTHING"). values are laid out row-major: [r0c0, r0c1, r1c0, r1c1, ...]. d selects the engine (placeholder style); dialect.Postgres{} and dialect.SQLite{} are both supported.
For upserts prefer BulkUpsert, which builds the conflict clause for you.
The query is built with "?" markers and rebound to the driver's bind style via db.Rebind (sqlx), so it is engine-portable: Postgres gets "$1,$2,…", SQLite/ MySQL keep "?". No per-dialect placeholder handling is needed.
func BulkUpsert ¶
func BulkUpsert(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, table string, columns []string, values []any, conflictColumns []string) error
BulkUpsert inserts rows and, on conflict over conflictColumns, updates every non-conflict column from the proposed row. The ON CONFLICT … DO UPDATE SET … = excluded.… form is understood by both Postgres and SQLite (3.24+); the query is rebound to the driver's placeholder style by BulkInsert.
func EnsureSchema ¶ added in v0.4.0
func EnsureSchema(ctx context.Context, log *logger.Logger, db *sqlx.DB, lockKey int64, ddl string) error
EnsureSchema applies idempotent DDL once, safely, at service startup. It runs ddl inside a transaction holding a Postgres transaction-scoped advisory lock keyed by lockKey, so when several replicas boot together they serialize on the lock and concurrent CREATE ... IF NOT EXISTS statements do not race ("tuple concurrently updated" / duplicate-relation errors). The lock releases automatically when the transaction ends — commit, rollback, or a dropped connection — so it cannot leak.
This is how a service auto-provisions the tables an SDK storage package owns (outbox, queue, auditlog, translation, ...) at startup, without hand-writing a migration for them. lockKey must be unique per schema across the database; derive it with AdvisoryKey.
func ExecContext ¶
ExecContext runs a parameterless statement.
func ExtContext ¶
func ExtContext(tx CommitRollbacker) (sqlx.ExtContext, error)
ExtContext extracts the sqlx.ExtContext (the query surface) from a transaction returned by Begin.
func NamedExecContext ¶
func NamedExecContext(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, data any) error
NamedExecContext runs an INSERT/UPDATE/DELETE with named parameters bound from data. It logs the expanded query and translates well-known Postgres errors.
func NamedExecContextRowsAffected ¶
func NamedExecContextRowsAffected(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, data any) (int64, error)
NamedExecContextRowsAffected is like NamedExecContext but returns the number of affected rows.
func NamedQuerySlice ¶
func NamedQuerySlice[T any](ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, data any, dest *[]T) error
NamedQuerySlice runs a named query and scans all rows into *[]T.
func NamedQuerySliceUsingIn ¶
func NamedQuerySliceUsingIn[T any](ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, data any, dest *[]T) error
NamedQuerySliceUsingIn runs a named query whose data contains a slice bound to an IN clause, scanning all rows into *[]T. It expands the slice parameters (sqlx.In) and rebinds the query for the driver before executing. Use it instead of NamedQuerySlice when the query has an `IN (:ids)` form.
func NamedQueryStruct ¶
func NamedQueryStruct(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, data, dest any) error
NamedQueryStruct runs a named query expected to return exactly one row.
func NamedQueryStructUsingIn ¶
func NamedQueryStructUsingIn(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, data, dest any) error
NamedQueryStructUsingIn runs a named query with an IN clause expected to return exactly one row, scanning it into dest. Returns ErrDBNotFound when no row matches.
func Open ¶
Open opens a sqlx.DB using the pgx driver. It does not verify connectivity; call StatusCheck for that.
func OpenSQLite ¶ added in v0.6.0
func OpenSQLite(cfg SQLiteConfig) (*sqlx.DB, error)
OpenSQLite opens a sqlx.DB for SQLite using a caller-registered driver (default "sqlite" = modernc.org/sqlite). It does not verify connectivity; call StatusCheck for that. All the dbx query helpers work against the returned handle unchanged — BulkInsert/BulkUpsert rebind placeholders to the SQLite bind style via db.Rebind.
func QuerySlice ¶
func QuerySlice[T any](ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, dest *[]T) error
QuerySlice runs a parameterless query and scans all rows into *[]T.
func QueryStruct ¶
func QueryStruct(ctx context.Context, log *logger.Logger, db sqlx.ExtContext, query string, dest any) error
QueryStruct runs a parameterless query expected to return exactly one row and scans it into dest (a pointer to a struct).
func StatusCheck ¶
StatusCheck pings the database, retrying until it is reachable or ctx is done.
On timeout it returns errors.Join(ctx.Err(), lastPingErr) rather than a bare "context deadline exceeded", so the real cause — connection refused, TLS failure, password authentication failed — is visible at the call site instead of being masked by the deadline.
If ctx has no deadline, DefaultStatusCheckTimeout is applied as a backstop so a persistent failure surfaces its error instead of hanging forever.
Types ¶
type Beginner ¶
type Beginner interface {
Begin() (CommitRollbacker, error)
}
Beginner starts a transaction. It is the seam stores depend on so they can be driven either by a pool or by an outer transaction (see middleware that opens a transaction per request).
type CommitRollbacker ¶
CommitRollbacker is a transaction that can be committed or rolled back.
type Config ¶
type Config struct {
User string
Password string
Host string
Name string
Schema string
MaxIdleConns int
MaxOpenConns int
DisableTLS bool
}
Config holds connection settings.
type DBBeginner ¶
type DBBeginner struct {
// contains filtered or unexported fields
}
DBBeginner adapts a *sqlx.DB to the Beginner interface.
func NewBeginner ¶
func NewBeginner(db *sqlx.DB) *DBBeginner
NewBeginner returns a Beginner backed by db.
func (*DBBeginner) Begin ¶
func (b *DBBeginner) Begin() (CommitRollbacker, error)
Begin starts a new transaction.
type JSONSlice ¶ added in v0.6.0
type JSONSlice[T any] []T
JSONSlice adapts a Go slice to a Postgres JSONB column. skit runs pgx v5 in database/sql mode (via sqlx), where database/sql cannot scan a Postgres array (text[], int8[]) back into a Go slice — there is no sql.Scanner for a bare slice, so an array-valued column fails StructScan. Storing the slice as JSONB and (un)marshaling it here sidesteps that: a JSONSlice[T] field StructScans cleanly and binds as a named parameter, so the store still uses one row struct.
Use it as the db-model field type for any slice-valued column:
type dbAdvert struct {
Photos dbx.JSONSlice[string] `db:"photos"`
}
backed by a column declared `photos JSONB NOT NULL DEFAULT '[]'::jsonb`.
type SQLiteConfig ¶ added in v0.6.0
type SQLiteConfig struct {
// Path is the database file (or ":memory:").
Path string
// Driver is the registered database/sql driver name; empty defaults to
// "sqlite" (modernc.org/sqlite, pure Go). The caller blank-imports the driver
// — dbx adds no SQLite driver dependency of its own.
Driver string
// Pragmas applied via the modernc DSN (_pragma=…); ignored when DSN is set.
JournalMode string // e.g. "WAL"; empty keeps the engine default
BusyTimeout time.Duration // busy_timeout; zero keeps the engine default
ForeignKeys bool // foreign_keys(1)
// DSN, when set, is used verbatim as the driver DSN — for drivers whose DSN
// syntax differs from modernc (e.g. mattn/go-sqlite3).
DSN string
MaxOpenConns int
MaxIdleConns int
}
SQLiteConfig configures a SQLite database opened by OpenSQLite.
Source Files
Directories