PostgreSQL con pgx y pgxpool

La biblioteca pgx es el driver de referencia para PostgreSQL en el ecosistema Go. En su versión 5 ofrece dos modos de uso: como driver compatible con database/sql y como API nativa. El modo nativo aprovecha características propias del protocolo de PostgreSQL (parámetros tipados, notificaciones asíncronas, COPY binario) que database/sql no expone. Esta lección se centra en el modo nativo, que es la opción recomendada para servicios nuevos.

Instalación y conexión básica

Para incorporar pgx al proyecto basta con añadir la dependencia:

go get github.com/jackc/pgx/v5
go get github.com/jackc/pgx/v5/pgxpool

La conexión al servidor se establece con una URL de PostgreSQL habitual:

package main

import (
    "context"
    "log/slog"
    "os"

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

func main() {
    dsn := os.Getenv("DATABASE_URL")
    // ejemplo: postgres://user:pass@localhost:5432/tienda?sslmode=disable

    pool, err := pgxpool.New(context.Background(), dsn)
    if err != nil {
        slog.Error("conectar pool", "err", err)
        os.Exit(1)
    }
    defer pool.Close()

    if err := pool.Ping(context.Background()); err != nil {
        slog.Error("ping", "err", err)
        os.Exit(1)
    }
    slog.Info("base de datos lista")
}

El tipo pgxpool.Pool representa un pool de conexiones mantenido por la librería. A diferencia del pool de database/sql, este está diseñado específicamente para PostgreSQL y expone opciones de configuración más ricas, como el healthcheck periódico o el tiempo máximo de vida de una conexión.

Configurar el pool con precisión

La mayoría de parámetros se definen construyendo primero una *pgxpool.Config a partir del DSN y ajustando los campos antes de crear el pool:

cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
    return nil, err
}

cfg.MaxConns = 25
cfg.MinConns = 2
cfg.MaxConnLifetime = 30 * time.Minute
cfg.MaxConnIdleTime = 5 * time.Minute
cfg.HealthCheckPeriod = 1 * time.Minute

pool, err := pgxpool.NewWithConfig(context.Background(), cfg)
if err != nil {
    return nil, err
}

Los valores concretos dependen del tipo de carga. Un servicio HTTP con tráfico sostenido se beneficia de MaxConns elevado y MinConns distinto de cero para evitar latencia de cold start. Un worker con ráfagas puntuales puede conformarse con un pool más modesto.

Mantén el pool como singleton en la aplicación. Pasa un puntero a los repositorios y componentes que lo necesiten. Crear y destruir pools por cada operación anula todas las optimizaciones de reuso.

Ejecutar consultas con parámetros nativos

En pgx los parámetros se indican con $1, $2, $3 siguiendo la convención de PostgreSQL. No se usa interpolación de cadenas en ningún caso porque abre la puerta a inyecciones SQL:

type Usuario struct {
    ID    int64
    Email string
    Alta  time.Time
}

func (r *Repo) BuscarPorEmail(ctx context.Context, email string) (*Usuario, error) {
    row := r.pool.QueryRow(ctx,
        "SELECT id, email, alta FROM usuarios WHERE email = $1",
        email,
    )
    var u Usuario
    if err := row.Scan(&u.ID, &u.Email, &u.Alta); err != nil {
        if errors.Is(err, pgx.ErrNoRows) {
            return nil, fmt.Errorf("usuario %s: %w", email, ErrNoEncontrado)
        }
        return nil, fmt.Errorf("scan usuario %s: %w", email, err)
    }
    return &u, nil
}

QueryRow es la forma idiomática para consultas que esperan cero o una fila. Si la consulta devuelve varias, usamos Query junto a un bucle for rows.Next() o una utilidad de escaneo colectivo.

Listar resultados con escaneo tipado

Una de las mayores comodidades de pgx es la familia de funciones pgx.RowTo y pgx.CollectRows. Permiten convertir filas en estructuras sin escribir código repetitivo:

func (r *Repo) Ultimos(ctx context.Context, n int) ([]Usuario, error) {
    rows, err := r.pool.Query(ctx,
        "SELECT id, email, alta FROM usuarios ORDER BY alta DESC LIMIT $1",
        n,
    )
    if err != nil {
        return nil, fmt.Errorf("query últimos: %w", err)
    }
    defer rows.Close()

    return pgx.CollectRows(rows, pgx.RowToStructByName[Usuario])
}

RowToStructByName empareja las columnas devueltas con los campos del struct por nombre (ignorando mayúsculas/minúsculas) o con el tag db si lo declaras. Para casos con campos renombrados:

type Usuario struct {
    ID    int64     `db:"id"`
    Email string    `db:"email"`
    Alta  time.Time `db:"alta"`
}

Este patrón elimina el clásico for rows.Next() { rows.Scan(&...) } que siempre es susceptible de errores de orden entre columnas y campos.

Insertar, actualizar y borrar

Para sentencias que no devuelven filas (INSERT, UPDATE, DELETE) se usa Exec. Devuelve un pgconn.CommandTag con el número de filas afectadas:

func (r *Repo) Actualizar(ctx context.Context, u Usuario) error {
    tag, err := r.pool.Exec(ctx,
        "UPDATE usuarios SET email = $1 WHERE id = $2",
        u.Email, u.ID,
    )
    if err != nil {
        return fmt.Errorf("update usuario %d: %w", u.ID, err)
    }
    if tag.RowsAffected() == 0 {
        return fmt.Errorf("update usuario %d: %w", u.ID, ErrNoEncontrado)
    }
    return nil
}

Cuando la operación es un INSERT con generación de ID por secuencia, el patrón idiomático es usar QueryRow con RETURNING id:

func (r *Repo) Crear(ctx context.Context, u *Usuario) error {
    row := r.pool.QueryRow(ctx,
        "INSERT INTO usuarios (email, alta) VALUES ($1, $2) RETURNING id",
        u.Email, time.Now(),
    )
    if err := row.Scan(&u.ID); err != nil {
        return fmt.Errorf("insert usuario %s: %w", u.Email, err)
    }
    return nil
}

Transacciones con Begin y Commit

Las transacciones en pgx se crean con BeginTx y se cierran con Commit o Rollback. La forma más segura utiliza defer para un rollback por precaución cuando no hay commit explícito:

func (r *Repo) Transferir(ctx context.Context, origen, destino int64, importe int64) error {
    tx, err := r.pool.Begin(ctx)
    if err != nil {
        return fmt.Errorf("begin: %w", err)
    }
    defer tx.Rollback(ctx) // ignorado si ya hicimos Commit

    if _, err := tx.Exec(ctx, "UPDATE cuentas SET saldo = saldo - $1 WHERE id = $2", importe, origen); err != nil {
        return fmt.Errorf("cargo origen: %w", err)
    }
    if _, err := tx.Exec(ctx, "UPDATE cuentas SET saldo = saldo + $1 WHERE id = $2", importe, destino); err != nil {
        return fmt.Errorf("abono destino: %w", err)
    }
    if err := tx.Commit(ctx); err != nil {
        return fmt.Errorf("commit: %w", err)
    }
    return nil
}

Pgx también ofrece pgx.BeginTxFunc como alternativa funcional que invoca una función dentro de una transacción y gestiona el rollback automático en caso de error:

err := pgx.BeginTxFunc(ctx, r.pool, pgx.TxOptions{}, func(tx pgx.Tx) error {
    if _, err := tx.Exec(ctx, sql1, args1...); err != nil {
        return err
    }
    _, err := tx.Exec(ctx, sql2, args2...)
    return err
})

Las transacciones sostienen una conexión del pool hasta su cierre. No las mantengas abiertas durante cálculos largos en memoria: empieza la transacción justo antes del trabajo contra la base de datos y ciérrala de inmediato.

flowchart LR
    A[pgxpool.Pool] --> B[Acquire conexión]
    B --> C[Query / Exec]
    C --> D[Release al pool]
    B --> E[Begin TX]
    E --> F[Exec dentro de TX]
    F --> G[Commit o Rollback]
    G --> D

Cargas masivas con COPY

Cuando necesitas insertar miles de filas, el comando COPY de PostgreSQL es órdenes de magnitud más rápido que INSERT individual. Pgx lo expone con CopyFrom:

filas := [][]any{
    {"ana@acme.com", time.Now()},
    {"luis@acme.com", time.Now()},
    {"marta@acme.com", time.Now()},
}

insertadas, err := r.pool.CopyFrom(
    ctx,
    pgx.Identifier{"usuarios"},
    []string{"email", "alta"},
    pgx.CopyFromRows(filas),
)
if err != nil {
    return fmt.Errorf("copy: %w", err)
}
slog.Info("inserción masiva", "filas", insertadas)

Para cargar un volumen grande leído desde un fichero o una API se implementa la interfaz pgx.CopyFromSource, evitando cargar todo en memoria.

Patrón repositorio con tests

La integración con context.Context en todas las llamadas facilita cancelar consultas cuando la petición del cliente termina. En los tests se combina pgx con un PostgreSQL efímero lanzado mediante testcontainers-go, obteniendo una base de datos limpia por cada suite sin mocks sintéticos:

func TestCrearUsuario(t *testing.T) {
    ctx := context.Background()
    pool := arrancarPostgresDeTest(t, ctx) // levanta contenedor, aplica migraciones
    defer pool.Close()

    repo := &Repo{pool: pool}
    u := Usuario{Email: "ana@acme.com"}
    if err := repo.Crear(ctx, &u); err != nil {
        t.Fatalf("Crear: %v", err)
    }
    if u.ID == 0 {
        t.Fatalf("ID no generado")
    }
}

Este enfoque sustituye la tentación de mockear el pool. Los mocks de SQL están desfasados respecto al estado real de la base y ocultan errores típicos (conversiones de tipos, zonas horarias, constraints). Con un contenedor efímero, cada prueba tiene el mismo entorno que producción.

pgx vs database/sql

Elegir entre database/sql con el driver pgx/v5/stdlib y el API nativa de pgx se resume en:

• 1. database/sql es útil cuando tu aplicación puede cambiar de motor (PostgreSQL, MySQL, SQLite) sin reescribir capas enteras. Encaja con ORMs y librerías de queries agnósticas del motor.

• 2. pgx nativo es la opción idónea cuando PostgreSQL es la elección firme. Expone LISTEN/NOTIFY, tipos específicos (arrays, JSONB tipado, intervals), parámetros con tipos nativos y rendimiento superior.

Para servicios nuevos orientados a PostgreSQL, el modo nativo de pgx suele ser la elección preferida por la comunidad Go actual. Si tu proyecto ya usa database/sql, la migración es gradual: se mantiene la interfaz genérica mientras se incorporan servicios nuevos directamente con pgx nativo.