pg

Documentation

Overview

Package pg provides a comprehensive suite of tools for interacting with PostgreSQL databases.

The package offers several core functionalities:

• Migrate: Manages database schema updates and ensures the database is always in its latest state.
• RequireDB: Facilitates test isolation by providing a mechanism for creating and tearing down unique PostgreSQL databases for each test case.
• [Store] and Transaction: Offer higher-level abstractions built upon the pgx library, providing a consistent interface and error handling for database interactions.
• [AtomicRead], QueryOne, and QueryMany: Provide optimized helper functions for common database operations, enhancing performance and code readability.

Index

• Variables
• func Migrate(ctx context.Context, namespace string, postgresDSN string, dir fs.FS, ...) ([]string, error)
• func QueryMany[Model any](ctx context.Context, q queryier, sql string, args ...any) ([]*Model, error)
• func QueryOne[Model any](ctx context.Context, q queryier, sql string, args ...any) (*Model, error)
• func RequireDB(t testing.TB) (p *pgxpool.Pool, pgURI string)
• type Atomic
• type Transaction
  • func LazyTx(db interface{ ... }, castPgErr func(error) error) Transaction
  • func OpenTx(ctx context.Context, db interface{ ... }, castPgErr func(error) error) (Transaction, error)

Variables

var (
	// ErrPg is the root error for all pg package errors.
	ErrPg = errors.New("pg")

	// ErrNotFound is returned when an operation cannot be completed
	// because the executed query returns no results.
	ErrNotFound = fmt.Errorf("%w: not found", ErrPg)

	// ErrConflict is returned when an operation cannot be completed
	// because applying changes would result in conflicting state.
	//
	// This error represents the whole class 23 of the Postgres integrity
	// constraint violation.
	ErrConflict = fmt.Errorf("%w: conflict", ErrPg)

	// ErrInvalidData is returned when an operation cannot be completed
	// because the provided data is invalid and cannot be accepted.
	//
	// This error represents the whole class 22 of the Postgres data
	// exception.
	ErrInvalidData = fmt.Errorf("%w: invalid input data", ErrPg)
)

var (
	ErrMigration        = errors.New("migration")
	ErrMigrationsLocked = fmt.Errorf("%w: locked", ErrMigration)
	ErrDuplicate        = fmt.Errorf("%w: duplicate", ErrMigration)
)

Functions

func Migrate

func Migrate(ctx context.Context, namespace string, postgresDSN string, dir fs.FS, migrationsGlob string) ([]string, error)

Migrate brings the database state up to the latest version by ensuring all migration files are applied.

All located migration files are sorted by their file name. It is expected that each migration file name is prefixed with a zero-padded number. An example migrations directory might look like so:

00001_init.sql
00002_add_foo.sql
00003_import_bar.sql

It is not required from migration files to have their prefix number sequential, therefore using UNIX time when creating a migration file is also a valid strategy and might be easier to use when dealing with conflicts:

$ touch "$(date '+%s')_remove_bar.sql"
$ ls
1723637107_init.sql
1723637110_add_foo.sql
1723637115_remove_bar.sql

Migration will fail if two files have the same prefix - this is most likely an unresolved merge conflict and should be fixed by renaming files to explicitly ensure the right order.

Migrate function executes all migrations within a single transaction, so that all changes are atomic. This might be limiting as for what a migration can do, but it is a great trade for keeping state consistent.

Migrate is using advisory lock to ensure that no two migrations can be executed at the same time.

Namespace is used to allow multiple, unrelated migrations running on the same database. This functionality is especially helpful when using a single database for local development or testing.

func QueryMany

func QueryMany[Model any](ctx context.Context, q queryier, sql string, args ...any) ([]*Model, error)

QueryMany is a helper function that will execute given query expecting any number of results.

Assigning of query result to model fields is done by matching structure "db" field tags to result columns. Make sure to tag your model fields. It is safe to use 'SELECT *' in most cases.

func QueryOne

func QueryOne[Model any](ctx context.Context, q queryier, sql string, args ...any) (*Model, error)

QueryOne executes given SQL query expecting at least one row to be returned.

If there is no row to return, ErrNotFound is the result.

Assigning of query result to model fields is done by matching structure "db" field tags to result columns. Make sure to tag your model fields. It is safe to use 'SELECT *' in most cases.

func RequireDB

func RequireDB(t testing.TB) (p *pgxpool.Pool, pgURI string)

RequireDB create a temporary Postgres database, available for the duration of the test.

The database connection can be configured by setting the PG_TEST_URI environment variable. By default, system local Postgres database is used.

Any test relying on this functionality is skipped when the PG_TEST_SKIP environment is set to a positive bollean value.

Types

type Atomic

type Atomic interface {
	// Commit commits the transaction if this is a real transaction or
	// releases the savepoint if this is a pseudo nested transaction.
	Commit(ctx context.Context) error

	// Rollback rolls back the transaction if this is a real transaction or
	// rolls back to the savepoint if this is a pseudo nested transaction.
	Rollback(ctx context.Context) error
}

Atomic is represented by a database transaction (or a nested pseudo-transaction) that allows to either apply or revert all contained changes.

type Transaction

type Transaction interface {
	Atomic

	// Exec executes a single SQL query, that does not return any result.
	//
	// This method is implemented by pgx Exec method.
	Exec(ctx context.Context, sql string, arguments ...any) (commandTag pgconn.CommandTag, err error)
	// Query execytes a single SQL query, that expects at least a single
	// row result.
	//
	// This method is implemented by pgx Query method.
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	// QueryRow execytes a single SQL query, that expects zero or more
	// result rows.
	//
	// This method is implemented by pgx QueryRow method.
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

Transaction allows to execute SQL commands within a single transaction.

func LazyTx

func LazyTx(
	db interface {
		Begin(context.Context) (pgx.Tx, error)
	},
	castPgErr func(error) error,
) Transaction

LazyTx returns a Transaction implementation that delays opening the database transaction until the first call.

Caller should provide the castPgErr argument function that inspects database error and where feasible, replaces it with package scoped error to not leak out underlying database interface.

func OpenTx

func OpenTx(
	ctx context.Context,
	db interface {
		Begin(context.Context) (pgx.Tx, error)
	},
	castPgErr func(error) error,
) (Transaction, error)

OpenTx opens a database transaction and returns the representing it Transaction instance.

Caller should provide the castPgErr argument function that inspects database error and where feasible, replaces it with package scoped error to not leak out underlying database interface.