Persistence

go.putnami.dev/sql provides PostgreSQL integration with connection pooling (via pgx/v5), transactions, a generic repository pattern, migrations, and a query builder.

Connection pool

Creating a pool

import "go.putnami.dev/sql"

pool, err := sql.NewPool(ctx, sql.PoolConfig{
    DSN:      "postgres://user:pass@localhost:5432/mydb?sslmode=disable",
    MaxConns: 10,
    MinConns: 2,
})
if err != nil {
    log.Fatal(err)
}
defer pool.Close()

Pool configuration

Field	Type	Default	Description
DSN	string	—	PostgreSQL connection string
MaxConns	int32	10	Maximum connections
MinConns	int32	2	Minimum idle connections
MaxConnLifetime	time.Duration	1h	Maximum connection lifetime
MaxConnIdleTime	time.Duration	30m	Maximum idle time
HealthCheckPeriod	time.Duration	1m	Health check interval

Direct queries

// Single row
row := pool.QueryRow(ctx, "SELECT name FROM users WHERE id = $1", userID)

// Multiple rows
rows, err := pool.Query(ctx, "SELECT * FROM users WHERE active = $1", true)

// Execute (INSERT, UPDATE, DELETE)
tag, err := pool.Exec(ctx, "DELETE FROM sessions WHERE expires_at < $1", time.Now())

// Health check
err := pool.Ping(ctx)

// Pool statistics
stats := pool.Stats()

Repository pattern

The generic Repository[T] provides typed CRUD operations for a database table:

Defining a repository

import (
    "github.com/jackc/pgx/v5"
    "go.putnami.dev/sql"
)

type User struct {
    ID    string
    Name  string
    Email string
    Age   int
}

func scanUser(row pgx.Row) (User, error) {
    var u User
    err := row.Scan(&u.ID, &u.Name, &u.Email, &u.Age)
    return u, err
}

type UserRepository struct {
    *sql.Repository[User]
}

func NewUserRepository(pool *sql.Pool) *UserRepository {
    return &UserRepository{
        Repository: sql.NewRepository[User](pool, "users", scanUser),
    }
}

Repository methods

repo := NewUserRepository(pool)

// Find by primary key
user, err := repo.FindByID(ctx, "id", "user-123")

// Find all
users, err := repo.FindAll(ctx)

// Find with WHERE clause
users, err := repo.FindWhere(ctx, "age > $1 AND active = $2", 18, true)

// Find one with WHERE clause
user, err := repo.FindOneWhere(ctx, "email = $1", "jane@example.com")

// Count
count, err := repo.Count(ctx, "active = $1", true)

// Exists
exists, err := repo.Exists(ctx, "email = $1", "jane@example.com")

// Delete
deleted, err := repo.DeleteWhere(ctx, "active = $1", false)
err = repo.DeleteByID(ctx, "id", "user-123")

// Raw queries
users, err := repo.Query(ctx, "SELECT * FROM users ORDER BY created_at DESC LIMIT $1", 10)
user, err := repo.QueryOne(ctx, "SELECT * FROM users WHERE email = $1", "jane@example.com")

Custom repository methods

Extend the repository with domain-specific queries:

func (r *UserRepository) FindByEmail(ctx context.Context, email string) (User, error) {
    return r.FindOneWhere(ctx, "email = $1", email)
}

func (r *UserRepository) FindActive(ctx context.Context) ([]User, error) {
    return r.FindWhere(ctx, "active = true")
}

func (r *UserRepository) Create(ctx context.Context, user User) error {
    q := r.Pool().Querier(ctx)
    _, err := q.Exec(ctx,
        "INSERT INTO users (id, name, email, age) VALUES ($1, $2, $3, $4)",
        user.ID, user.Name, user.Email, user.Age,
    )
    return err
}

Transactions

Basic transaction

err := sql.WithTx(ctx, pool, func(ctx context.Context) error {
    // All queries within this function use the same transaction
    _, err := pool.Exec(ctx, "INSERT INTO users (id, name) VALUES ($1, $2)", id, name)
    if err != nil {
        return err // triggers rollback
    }

    _, err = pool.Exec(ctx, "INSERT INTO audit_log (user_id, action) VALUES ($1, $2)", id, "created")
    return err // nil = commit, error = rollback
})

Nested transactions

Nested calls to WithTx reuse the existing transaction:

err := sql.WithTx(ctx, pool, func(ctx context.Context) error {
    repo.Create(ctx, user)

    // This is still the same transaction
    return sql.WithTx(ctx, pool, func(ctx context.Context) error {
        return repo.CreateAuditLog(ctx, user.ID, "created")
    })
})

Transaction context

The pool's Querier method automatically detects transactions from the context:

// Outside a transaction: uses pool connection
q := pool.Querier(ctx)

// Inside WithTx: uses the transaction
sql.WithTx(ctx, pool, func(ctx context.Context) error {
    q := pool.Querier(ctx) // uses the transaction connection
    q.Exec(ctx, "...")
    return nil
})

You can also check for an active transaction:

tx := sql.TxFromContext(ctx)
if tx != nil {
    // inside a transaction
}

Query builder

A lightweight builder for common SQL patterns:

SELECT

import "go.putnami.dev/sql"

query, args := sql.Select("users").
    Columns("id", "name", "email").
    Where("age > $1", 18).
    Where("active = $2", true).
    OrderBy("name ASC").
    Limit(10).
    Offset(20).
    Build()

// query: "SELECT id, name, email FROM users WHERE age > $1 AND active = $2 ORDER BY name ASC LIMIT 10 OFFSET 20"
// args: [18, true]

INSERT

query, args := sql.Insert("users").
    Columns("id", "name", "email").
    Values("user-123", "Jane", "jane@example.com").
    Returning("id", "created_at").
    Build()

UPDATE

query, args := sql.Update("users").
    Set("name = $1", "Jane Doe").
    Set("updated_at = $2", time.Now()).
    Where("id = $3", "user-123").
    Build()

DELETE

query, args := sql.Delete("users").
    Where("active = $1", false).
    Returning("id").
    Build()

Migrations

Database schema migrations via goose:

import "go.putnami.dev/sql"

migrator := sql.NewMigrator(db, sql.MigrationConfig{
    Dir: "migrations",
})

// Run pending migrations
err := migrator.Up(ctx)

// Roll back the last migration
err := migrator.Down(ctx)

// Print migration status
err := migrator.Status(ctx)

// Get current version
version, err := migrator.Version(ctx)

Plugin

The SQL plugin integrates the connection pool with the application lifecycle:

import (
    "go.putnami.dev/app"
    "go.putnami.dev/sql"
)

a := app.New("my-service")
a.Module.Use(sql.NewPlugin(sql.PluginConfig{
    DSN:      "postgres://localhost/mydb",
    MaxConns: 10,
}))

The plugin:

• Creates the connection pool during warmup
• Registers *sql.Pool in the DI container
• Closes the pool on application shutdown

Error codes

Code	Description
db.connection	Connection pool creation failed
db.query	Query execution failed
db.migration	Migration failed
db.transaction	Transaction begin/commit/rollback failed

Related guides

• Dependency Injection — injecting the pool
• Plugins & Lifecycle — SQL plugin lifecycle
• Errors — structured error handling