pgxkit

pgxkit

A production-ready PostgreSQL toolkit for Go applications — tool-agnostic utilities for connection pooling, observability, testing, and type helpers.

Overview

pgxkit is a tool-agnostic PostgreSQL toolkit that works with any approach to PostgreSQL development:

• Raw pgx usage - Use pgxkit directly with pgx for maximum control
• Any code generation tool - Works with sqlc, Skimatik, or any other tool
• Clean architecture - Separate your database layer from your business logic
• Production-ready - Built-in observability, retry logic, and graceful shutdown

Key Features

• 🔄 Read/Write Pool Abstraction - Safe by default, optimized when needed
• 🎣 Extensible Hook System - Add logging, tracing, metrics, circuit breakers
• 🔁 Smart Retry Logic - PostgreSQL-aware error detection and exponential backoff
• 🧪 Testing Infrastructure - Golden test support for performance regression detection
• 🔧 Type Helpers - Seamless pgx type conversions
• 📊 Health Checks - Built-in database connectivity monitoring
• 🛡️ Graceful Shutdown - Production-ready lifecycle management
• 🔀 Executor Interface - Write functions that work with both *DB and *Tx

Installation

go get github.com/nhalm/pgxkit

Quick Start

Basic Usage

package main

import (
    "context"
    "log"

    "github.com/nhalm/pgxkit"
)

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

    // Create and connect to database
    db := pgxkit.NewDB()

    // Connect using environment variables or explicit DSN
    err := db.Connect(ctx, "") // Uses POSTGRES_* env vars
    if err != nil {
        log.Fatal(err)
    }
    defer db.Shutdown(ctx)

    // Execute queries (uses write pool by default - safe)
    _, err = db.Exec(ctx, "INSERT INTO users (name) VALUES ($1)", "John")
    if err != nil {
        log.Fatal(err)
    }

    // Optimize reads with explicit read pool usage
    rows, err := db.ReadQuery(ctx, "SELECT id, name FROM users")
    if err != nil {
        log.Fatal(err)
    }
    defer rows.Close()

    // Process results...
}

With Read/Write Splitting

db := pgxkit.NewDB()

// Connect with separate read and write pools
err := db.ConnectReadWrite(ctx, readDSN, writeDSN)
if err != nil {
    log.Fatal(err)
}

// Writes go to write pool (safe by default)
db.Exec(ctx, "INSERT INTO users ...")

// Reads can use read pool for optimization
db.ReadQuery(ctx, "SELECT * FROM users")

Configuration

Environment Variables

pgxkit uses these environment variables when no DSN is provided:

• POSTGRES_HOST (default: "localhost")
• POSTGRES_PORT (default: 5432)
• POSTGRES_USER (default: "postgres")
• POSTGRES_PASSWORD (default: "")
• POSTGRES_DB (default: "postgres")
• POSTGRES_SSLMODE (default: "disable")

DSN Utilities

// Get DSN from environment variables
dsn := pgxkit.GetDSN()

// Use with external tools
dsn := pgxkit.GetDSN()

Hooks System

Add observability and custom functionality through connect options:

db := pgxkit.NewDB()
err := db.Connect(ctx, dsn,
    // Logging hook
    pgxkit.WithBeforeOperation(func(ctx context.Context, sql string, args []interface{}, operationErr error) error {
        log.Printf("Executing: %s", sql)
        return nil
    }),
    // Metrics hook
    pgxkit.WithAfterOperation(func(ctx context.Context, sql string, args []interface{}, operationErr error) error {
        if operationErr != nil {
            metrics.IncrementCounter("db.errors")
        }
        return nil
    }),
    // Connection setup
    pgxkit.WithOnConnect(func(conn *pgx.Conn) error {
        log.Println("New connection established")
        return nil
    }),
)

Available Options

Pool configuration:

• WithMaxConns(n int32) - Maximum connections in pool
• WithMinConns(n int32) - Minimum connections in pool
• WithMaxConnLifetime(d time.Duration) - Maximum connection lifetime
• WithMaxConnIdleTime(d time.Duration) - Maximum idle time

Operation hooks:

• WithBeforeOperation(fn) - Before any query/exec
• WithAfterOperation(fn) - After any query/exec
• WithBeforeTransaction(fn) - Before transaction starts
• WithAfterTransaction(fn) - After transaction completes
• WithOnShutdown(fn) - During graceful shutdown

Connection hooks:

• WithOnConnect(fn) - When new connection is established
• WithOnDisconnect(fn) - When connection is closed
• WithOnAcquire(fn) - When connection is acquired from pool
• WithOnRelease(fn) - When connection is returned to pool

Retry Logic

RetryOperation Function

// Retry with default settings:
// - 3 retry attempts
// - 100ms initial delay
// - 1s maximum delay
// - 2x exponential backoff
err := pgxkit.RetryOperation(ctx, func(ctx context.Context) error {
    _, err := db.Exec(ctx, "INSERT INTO users (name, email) VALUES ($1, $2)",
        "John Doe", "john@example.com")
    return err
})

// Retry with custom configuration
err = pgxkit.RetryOperation(ctx, func(ctx context.Context) error {
    rows, err := db.Query(ctx, "SELECT * FROM users")
    if err != nil {
        return err
    }
    defer rows.Close()
    // Process rows...
    return nil
}, pgxkit.WithMaxRetries(5), pgxkit.WithMaxDelay(5*time.Second))

// Retry with timeout using context.WithTimeout
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
user, err := pgxkit.Retry(ctx, func(ctx context.Context) (*User, error) {
    return getUserFromDatabase(ctx)
}, pgxkit.WithMaxRetries(3))

Timeout Behavior

The timeout (set via context.WithTimeout) applies to all retry attempts combined, not per-attempt. If your timeout is 5 seconds and the first attempt takes 3 seconds, subsequent retries share the remaining 2 seconds.

Smart Error Detection

pgxkit automatically detects which PostgreSQL errors are worth retrying:

// Check if an error would be retried
if pgxkit.IsRetryableError(err) {
    log.Println("Transient error - would retry")
} else {
    log.Println("Permanent error - would not retry")
}

Health Checks

// Check database connectivity
if db.IsReady(ctx) {
    log.Println("Database is ready")
}

// Detailed health check
if err := db.HealthCheck(ctx); err != nil {
    log.Printf("Database health check failed: %v", err)
}

Testing

Basic Testing

func TestUserOperations(t *testing.T) {
    testDB := pgxkit.NewTestDB()
    ctx := context.Background()
    err := testDB.Connect(ctx, "") // Uses TEST_DATABASE_URL env var
    if err != nil {
        t.Skip("Test database not available")
    }
    defer testDB.Shutdown(ctx)

    err = testDB.Setup()
    if err != nil {
        t.Skip("Test database setup failed")
    }
    defer testDB.Clean()

    // Use testDB.DB for your tests
    _, err = testDB.Exec(ctx, "INSERT INTO users ...")
    // ... test assertions
}

Golden Testing (Performance Regression Detection)

func TestUserQueries(t *testing.T) {
    testDB := pgxkit.NewTestDB()
    ctx := context.Background()
    err := testDB.Connect(ctx, "")
    if err != nil {
        t.Skip("Test database not available")
    }
    defer testDB.Shutdown(ctx)

    testDB.Setup()
    defer testDB.Clean()

    // Enable golden test hooks - captures EXPLAIN plans automatically
    db := testDB.EnableGolden("TestUserQueries")

    // These queries will have their EXPLAIN plans captured
    rows, err := db.Query(ctx, "SELECT * FROM users WHERE active = true")
    // ... more queries

    // Plans are saved to testdata/golden/TestUserQueries_query_1.json, etc.
    // Future runs compare plans to detect performance regressions
}

Type Helpers

Seamless conversions between Go types and pgx types:

// String conversions
pgxText := pgxkit.ToPgxText(&myString)
stringPtr := pgxkit.FromPgxText(pgxText)

// Numeric conversions
pgxInt := pgxkit.ToPgxInt8(&myInt64)
intPtr := pgxkit.FromPgxInt8(pgxInt)

// UUID conversions
pgxUUID := pgxkit.ToPgxUUID(myUUID)
uuid := pgxkit.FromPgxUUID(pgxUUID)

// Time conversions
pgxTime := pgxkit.ToPgxTimestamptz(&myTime)
timePtr := pgxkit.FromPgxTimestamptz(pgxTime)

// Array conversions
pgxArray := pgxkit.ToPgxTextArray(myStringSlice)
stringSlice := pgxkit.FromPgxTextArray(pgxArray)

Production Features

Graceful Shutdown

// Graceful shutdown with timeout
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

err := db.Shutdown(ctx)
if err != nil {
    log.Printf("Shutdown completed with warnings: %v", err)
}

Connection Statistics

// Get pool statistics
stats := db.Stats()          // Write pool stats
readStats := db.ReadStats()  // Read pool stats (if using read/write split)

log.Printf("Active connections: %d", stats.AcquiredConns())
log.Printf("Idle connections: %d", stats.IdleConns())

Error Handling

// Structured error types
err := pgxkit.NewNotFoundError("User", userID)
err := pgxkit.NewValidationError("Email", "create", "address", "invalid format", nil)
err := pgxkit.NewDatabaseError("Order", "query", originalErr)

// Type checking
var notFoundErr *pgxkit.NotFoundError
if errors.As(err, &notFoundErr) {
    // Handle not found error
}

Architecture

pgxkit follows these design principles:

1. Safety First - All default methods use write pool for consistency
2. Explicit Optimization - Use ReadQuery() methods for read optimization
3. Tool Agnostic - Works with any PostgreSQL development approach
4. Extensible - Hook system for custom functionality
5. Production Ready - Built-in observability and lifecycle management

Examples

With Raw pgx

db := pgxkit.NewDB()
db.Connect(ctx, "postgres://...")

// Use pgx directly with pgxkit utilities
rows, err := db.Query(ctx, "SELECT * FROM users")
for rows.Next() {
    var user User
    err := rows.Scan(&user.ID, &user.Name)
    // ...
}

With Any Code Generation Tool

// Works with sqlc, Skimatik, or any other tool
db := pgxkit.NewDB()
db.Connect(ctx, "postgres://...")

// Use your generated code with pgxkit's connection
queries := sqlc.New(db.GetPool()) // or your tool's constructor
users, err := queries.GetAllUsers(ctx)

With Hooks for Observability

db := pgxkit.NewDB()
err := db.Connect(ctx, dsn,
    // Add tracing
    pgxkit.WithBeforeOperation(func(ctx context.Context, sql string, args []interface{}, operationErr error) error {
        span := trace.SpanFromContext(ctx)
        span.SetAttributes(attribute.String("db.statement", sql))
        return nil
    }),
    // Add metrics
    pgxkit.WithAfterOperation(func(ctx context.Context, sql string, args []interface{}, operationErr error) error {
        status := "success"
        if operationErr != nil {
            status = "error"
        }
        queryCounter.WithLabelValues(status).Inc()
        return nil
    }),
)

Documentation

Visit the pgxkit Wiki - Complete documentation with examples and guides

Quick Links

• Getting Started - Setup and basic usage
• API Reference - Complete API documentation
• Examples - Practical code examples and use cases
• Performance Guide - Optimization strategies
• Production Guide - Deployment best practices
• Testing Guide - Testing strategies and golden tests
• FAQ - Frequently asked questions

Contributing

We welcome contributions! Please see our Contributing Guide for details.

License

MIT License - see LICENSE file for details.