Go Database Patterns with sqlx and pgx

Go’s standard database/sql package is solid but verbose. You spend a lot of time scanning columns into struct fields manually. Two libraries solve this elegantly: sqlx extends database/sql with struct scanning and named parameters, while pgx is a pure Go PostgreSQL driver that gives you maximum performance and PostgreSQL-specific features. This guide shows you how to use both effectively.

Getting Started with sqlx

Install sqlx and a PostgreSQL driver.

go get github.com/jmoiron/sqlx
go get github.com/lib/pq

Connect to the database and define your models with db struct tags.

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/jmoiron/sqlx"
	_ "github.com/lib/pq"
)

type User struct {
	ID        int       `db:"id"`
	Email     string    `db:"email"`
	Name      string    `db:"name"`
	CreatedAt time.Time `db:"created_at"`
	UpdatedAt time.Time `db:"updated_at"`
}

type Post struct {
	ID          int       `db:"id"`
	UserID      int       `db:"user_id"`
	Title       string    `db:"title"`
	Body        string    `db:"body"`
	PublishedAt *time.Time `db:"published_at"`
	CreatedAt   time.Time `db:"created_at"`
}

func main() {
	db, err := sqlx.Connect("postgres",
		"host=localhost port=5432 user=app dbname=myapp sslmode=disable")
	if err != nil {
		log.Fatalf("failed to connect: %v", err)
	}
	defer db.Close()

	// Configure connection pool
	db.SetMaxOpenConns(25)
	db.SetMaxIdleConns(5)
	db.SetConnMaxLifetime(5 * time.Minute)

	fmt.Println("Connected to database")
}

Querying with sqlx

sqlx eliminates manual column scanning. Use Get for single rows and Select for multiple rows.

// Get a single user
var user User
err := db.GetContext(ctx, &user, "SELECT * FROM users WHERE id = $1", userID)
if err != nil {
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrNotFound
	}
	return nil, fmt.Errorf("get user: %w", err)
}

// Get multiple users
var users []User
err := db.SelectContext(ctx, &users,
	"SELECT * FROM users WHERE created_at > $1 ORDER BY name", since)
if err != nil {
	return nil, fmt.Errorf("list users: %w", err)
}

// Named queries: use struct fields as parameters
user := User{Name: "Alice", Email: "alice@example.com"}
result, err := db.NamedExecContext(ctx,
	`INSERT INTO users (name, email, created_at, updated_at)
	 VALUES (:name, :email, NOW(), NOW())`, user)
if err != nil {
	return fmt.Errorf("insert user: %w", err)
}

The Repository Pattern with sqlx

Wrap database access behind a repository interface for clean separation.

package postgres

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	"github.com/jmoiron/sqlx"
	"myapp/internal/domain"
)

type UserRepository struct {
	db *sqlx.DB
}

func NewUserRepository(db *sqlx.DB) *UserRepository {
	return &UserRepository{db: db}
}

func (r *UserRepository) Create(ctx context.Context, user *domain.User) error {
	query := `
		INSERT INTO users (email, name, created_at, updated_at)
		VALUES ($1, $2, NOW(), NOW())
		RETURNING id, created_at, updated_at`

	return r.db.QueryRowxContext(ctx, query, user.Email, user.Name).
		Scan(&user.ID, &user.CreatedAt, &user.UpdatedAt)
}

func (r *UserRepository) GetByID(ctx context.Context, id int) (*domain.User, error) {
	var user domain.User
	err := r.db.GetContext(ctx, &user, "SELECT * FROM users WHERE id = $1", id)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, domain.ErrNotFound
		}
		return nil, fmt.Errorf("get user by id: %w", err)
	}
	return &user, nil
}

func (r *UserRepository) GetByEmail(ctx context.Context, email string) (*domain.User, error) {
	var user domain.User
	err := r.db.GetContext(ctx, &user, "SELECT * FROM users WHERE email = $1", email)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, domain.ErrNotFound
		}
		return nil, fmt.Errorf("get user by email: %w", err)
	}
	return &user, nil
}

func (r *UserRepository) Update(ctx context.Context, user *domain.User) error {
	query := `
		UPDATE users SET name = $1, email = $2, updated_at = NOW()
		WHERE id = $3
		RETURNING updated_at`

	err := r.db.QueryRowxContext(ctx, query, user.Name, user.Email, user.ID).
		Scan(&user.UpdatedAt)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return domain.ErrNotFound
		}
		return fmt.Errorf("update user: %w", err)
	}
	return nil
}

func (r *UserRepository) List(ctx context.Context, limit, offset int) ([]domain.User, error) {
	var users []domain.User
	err := r.db.SelectContext(ctx, &users,
		"SELECT * FROM users ORDER BY created_at DESC LIMIT $1 OFFSET $2",
		limit, offset)
	if err != nil {
		return nil, fmt.Errorf("list users: %w", err)
	}
	return users, nil
}

func (r *UserRepository) Delete(ctx context.Context, id int) error {
	result, err := r.db.ExecContext(ctx, "DELETE FROM users WHERE id = $1", id)
	if err != nil {
		return fmt.Errorf("delete user: %w", err)
	}
	rows, _ := result.RowsAffected()
	if rows == 0 {
		return domain.ErrNotFound
	}
	return nil
}

Transactions with sqlx

For operations that must succeed or fail together, use transactions.

func (r *OrderRepository) CreateOrder(ctx context.Context, order *domain.Order) error {
	tx, err := r.db.BeginTxx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin transaction: %w", err)
	}
	// Defer a rollback. If the transaction was committed, this is a no-op.
	defer tx.Rollback()

	// Insert the order
	err = tx.QueryRowxContext(ctx,
		`INSERT INTO orders (user_id, total, status, created_at)
		 VALUES ($1, $2, $3, NOW()) RETURNING id, created_at`,
		order.UserID, order.Total, "pending").
		Scan(&order.ID, &order.CreatedAt)
	if err != nil {
		return fmt.Errorf("insert order: %w", err)
	}

	// Insert order items
	for i := range order.Items {
		item := &order.Items[i]
		_, err := tx.ExecContext(ctx,
			`INSERT INTO order_items (order_id, product_id, quantity, price)
			 VALUES ($1, $2, $3, $4)`,
			order.ID, item.ProductID, item.Quantity, item.Price)
		if err != nil {
			return fmt.Errorf("insert order item: %w", err)
		}
	}

	// Deduct inventory
	for _, item := range order.Items {
		result, err := tx.ExecContext(ctx,
			`UPDATE products SET stock = stock - $1
			 WHERE id = $2 AND stock >= $1`,
			item.Quantity, item.ProductID)
		if err != nil {
			return fmt.Errorf("deduct stock: %w", err)
		}
		rows, _ := result.RowsAffected()
		if rows == 0 {
			return fmt.Errorf("insufficient stock for product %d", item.ProductID)
		}
	}

	return tx.Commit()
}

Using pgx Directly

pgx is a PostgreSQL-specific driver that offers better performance and PostgreSQL-specific features like COPY, LISTEN/NOTIFY, and custom types. Use pgx when you need these features or want maximum performance.

package main

import (
	"context"
	"fmt"
	"log"
	"os"

	"github.com/jackc/pgx/v5/pgxpool"
)

func main() {
	ctx := context.Background()

	pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		log.Fatalf("unable to create pool: %v", err)
	}
	defer pool.Close()

	// Simple query
	var name string
	var email string
	err = pool.QueryRow(ctx,
		"SELECT name, email FROM users WHERE id = $1", 1).
		Scan(&name, &email)
	if err != nil {
		log.Fatalf("query failed: %v", err)
	}
	fmt.Printf("User: %s <%s>\n", name, email)
}

Batch Operations with pgx

pgx supports batch queries that send multiple statements in a single round trip.

func (r *ProductRepo) CreateBatch(ctx context.Context, products []Product) error {
	batch := &pgx.Batch{}

	for _, p := range products {
		batch.Queue(
			"INSERT INTO products (name, price, stock) VALUES ($1, $2, $3)",
			p.Name, p.Price, p.Stock,
		)
	}

	results := r.pool.SendBatch(ctx, batch)
	defer results.Close()

	for range products {
		_, err := results.Exec()
		if err != nil {
			return fmt.Errorf("batch insert: %w", err)
		}
	}

	return nil
}

COPY Protocol for Bulk Inserts

For inserting thousands of rows, the PostgreSQL COPY protocol is much faster than individual INSERT statements. pgx supports this natively.

func (r *EventRepo) BulkInsert(ctx context.Context, events []Event) (int64, error) {
	rows := make([][]any, len(events))
	for i, e := range events {
		rows[i] = []any{e.Type, e.Payload, e.CreatedAt}
	}

	copyCount, err := r.pool.CopyFrom(
		ctx,
		pgx.Identifier{"events"},
		[]string{"type", "payload", "created_at"},
		pgx.CopyFromRows(rows),
	)
	if err != nil {
		return 0, fmt.Errorf("copy from: %w", err)
	}

	return copyCount, nil
}

LISTEN/NOTIFY for Real-Time Updates

pgx supports PostgreSQL’s LISTEN/NOTIFY for real-time event streaming between your application and the database.

func listenForChanges(ctx context.Context, pool *pgxpool.Pool) error {
	conn, err := pool.Acquire(ctx)
	if err != nil {
		return fmt.Errorf("acquire connection: %w", err)
	}
	defer conn.Release()

	_, err = conn.Exec(ctx, "LISTEN order_events")
	if err != nil {
		return fmt.Errorf("listen: %w", err)
	}

	for {
		notification, err := conn.Conn().WaitForNotification(ctx)
		if err != nil {
			return fmt.Errorf("wait for notification: %w", err)
		}

		fmt.Printf("Channel: %s, Payload: %s\n",
			notification.Channel, notification.Payload)
	}
}

Connection Pool Configuration

Both sqlx and pgx support connection pooling. Proper configuration prevents connection exhaustion.

// sqlx pool configuration
db.SetMaxOpenConns(25)              // Max connections in pool
db.SetMaxIdleConns(5)               // Max idle connections
db.SetConnMaxLifetime(5 * time.Minute) // Max connection age
db.SetConnMaxIdleTime(1 * time.Minute) // Max idle time

// pgx pool configuration
config, _ := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
config.MaxConns = 25
config.MinConns = 5
config.MaxConnLifetime = 5 * time.Minute
config.MaxConnIdleTime = 1 * time.Minute
config.HealthCheckPeriod = 30 * time.Second

pool, _ := pgxpool.NewWithConfig(ctx, config)

A good starting point is MaxConns equal to 2 * number_of_cpu_cores + number_of_disks on your database server. For cloud databases, check the provider’s connection limits.

Handling NULL Values

Database NULLs require pointer types or sql.Null types in Go.

// Using pointers (works with both sqlx and pgx)
type Article struct {
	ID          int        `db:"id"`
	Title       string     `db:"title"`
	PublishedAt *time.Time `db:"published_at"` // nil when NULL
	DeletedAt   *time.Time `db:"deleted_at"`
}

// Using sql.Null types
type Article struct {
	ID          int            `db:"id"`
	Title       string         `db:"title"`
	PublishedAt sql.NullTime   `db:"published_at"`
	ViewCount   sql.NullInt64  `db:"view_count"`
	Subtitle    sql.NullString `db:"subtitle"`
}

Pointer types are simpler. Use sql.Null* types when you need to distinguish between a zero value and NULL.

Testing Database Code

Use a test database and transactions that roll back after each test.

func setupTestDB(t *testing.T) *sqlx.DB {
	t.Helper()

	db, err := sqlx.Connect("postgres",
		"host=localhost dbname=myapp_test sslmode=disable")
	if err != nil {
		t.Fatalf("connect: %v", err)
	}

	t.Cleanup(func() { db.Close() })
	return db
}

func TestUserRepository_Create(t *testing.T) {
	db := setupTestDB(t)

	// Start a transaction that we will roll back
	tx, err := db.Beginx()
	if err != nil {
		t.Fatalf("begin: %v", err)
	}
	t.Cleanup(func() { tx.Rollback() })

	repo := NewUserRepository(tx) // Accept sqlx.ExtContext interface

	user := &domain.User{
		Name:  "Test User",
		Email: "test@example.com",
	}

	err = repo.Create(context.Background(), user)
	if err != nil {
		t.Fatalf("create: %v", err)
	}

	if user.ID == 0 {
		t.Error("expected ID to be set")
	}

	// Verify the user exists within the transaction
	got, err := repo.GetByID(context.Background(), user.ID)
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if got.Email != "test@example.com" {
		t.Errorf("email = %q, want %q", got.Email, "test@example.com")
	}
}

Wrapping Up

sqlx and pgx solve different problems in Go database access. sqlx extends database/sql with convenient struct scanning, named parameters, and In clause expansion, working with any SQL database. pgx gives you PostgreSQL-specific power: COPY protocol for bulk loads, LISTEN/NOTIFY for real-time updates, and batch queries for reduced round trips. Use sqlx when you want portability and simplicity. Use pgx when you are committed to PostgreSQL and need its advanced features. In both cases, wrap your database access behind repository interfaces, handle transactions properly, and configure your connection pool to match your workload.