peque ~dev-fx

Peque — PostgreSQL client for D

Peque is a lightweight libpq wrapper for the D programming language.

It uses SafeRefCounted (from std.typecons) to manage libpq objects deterministically — connections and results are freed as soon as they go out of scope, without depending on the GC.

Features

• Reference-counted Connection and Result — deterministic cleanup, no GC dependency
• Parameterized queries via execParams — SQL injection safe by design
• Automatic bidirectional type conversion between PostgreSQL text format and D types
• Nullable!T support for NULL columns
• RAII transaction helper with auto-commit / auto-rollback on exception
• Configurable transaction isolation level (readCommitted, repeatableRead, serializable, serverDefault)
• Savepoint support for partial rollbacks within a transaction
• Static or dynamic (bindbc-loader) loading of libpq

Supported types

Installation

dub.json:

"dependencies": {
    "peque": "~>0.1.0"
}

dub.sdl:

dependency "peque" version="~>0.1.0"

Choose a configuration depending on how you want to link libpq:

Quick start

import peque;
import std.stdio;

auto c = Connection(
    dbname: "mydb",
    user:   "myuser",
    password: "secret",
    host:   "localhost",
    port:   "5432",
);

// Raw SQL — fine for DDL or trusted input
c.exec("
    CREATE TABLE IF NOT EXISTS items (id serial, name text, qty int);
    INSERT INTO items (name, qty) VALUES ('apple', 5), ('banana', 3);
");

// Parameterized query — SQL-injection safe
auto res = c.execParams(
    "SELECT name, qty FROM items WHERE qty > $1", 2);

foreach (row; res)
    writeln(row["name"].get!string, ": ", row["qty"].get!int);

// Nullable column
auto maybeQty = res[0]["qty"].get!(Nullable!int);
if (!maybeQty.isNull)
    writeln(maybeQty.get);

Prepared statements

Connection.prepare() registers a server-side prepared statement and returns a move-only PreparedStatement handle. Its destructor issues DEALLOCATE automatically when it goes out of scope.

auto stmt = conn.prepare("find_user",
    "SELECT id, name FROM users WHERE id = $1");

auto result = stmt.exec(42);
// stmt goes out of scope → DEALLOCATE find_user sent automatically

Useful when the same query runs many times in the same connection session — the server parses and plans it once.

Transactions

Connection.transaction() runs a delegate inside a BEGIN/COMMIT block. On exception the transaction is always rolled back. The delegate receives a ref Transaction handle that exposes exec, execParams, and escapeString but intentionally hides commit() and rollback() — making accidental early termination of the transaction a compile-time error rather than a silent runtime bug.

c.transaction((ref tx) {
    tx.execParams("INSERT INTO items (name, qty) VALUES ($1, $2)", "cherry", 10);
    tx.execParams("UPDATE items SET qty = qty - 1 WHERE name = $1", "apple");
});

transaction() can return a value:

auto newQty = c.transaction((ref tx) {
    tx.execParams("UPDATE items SET qty = qty - 1 WHERE name = $1", "apple");
    return tx.execParams(
        "SELECT qty FROM items WHERE name = $1", "apple")[0][0].get!int;
});

Use OnSuccess.rollback for dry-runs or test helpers that must not persist changes:

c.transaction!(OnSuccess.rollback)((ref tx) {
    tx.execParams("DELETE FROM items");
    auto count = tx.execParams("SELECT count(*) FROM items")[0][0].get!long;
    assert(count == 0);
    // transaction is rolled back after the delegate returns
});

Business logic functions take ref Transaction and are completely unaware of nesting depth:

void deductStock(ref Transaction tx, string item, int qty) {
    tx.execParams(
        "UPDATE items SET qty = qty - $1 WHERE name = $2", qty, item);
}

c.transaction((ref tx) { deductStock(tx, "apple", 1); });

Savepoints

Transaction.savepoint() creates a PostgreSQL savepoint. On exception, only the savepoint changes are rolled back — the enclosing transaction remains open and intact. The delegate receives the same ref Transaction, so the same business logic functions work at any nesting depth without modification.

c.transaction((ref tx) {
    tx.execParams("INSERT INTO items (name, qty) VALUES ($1, $2)", "date", 7);

    try {
        tx.savepoint((ref tx) {
            tx.execParams(
                "INSERT INTO items (name, qty) VALUES ($1, $2)", "elderberry", 2);
            throw new Exception("changed my mind");
            // only the elderberry insert is rolled back
        });
    } catch (Exception e) {}

    // date is still in the transaction and will be committed
});

Savepoints can be nested arbitrarily. OnSuccess.rollback works on savepoints too — useful for dry-run sub-operations inside a larger transaction.

Isolation levels

The IsolationLevel template parameter controls the PostgreSQL transaction isolation level. The default is readCommitted, which is always set explicitly in the BEGIN statement regardless of server configuration — so the behaviour is predictable even if the server or role has a different default_transaction_isolation configured.

import peque;

// Read committed (default) — explicit in BEGIN, immune to server config
c.transaction((ref tx) { ... });

// Repeatable read
c.transaction!(OnSuccess.commit, IsolationLevel.repeatableRead)((ref tx) { ... });

// Serializable — may throw on conflict, application must be prepared to retry
c.transaction!(OnSuccess.commit, IsolationLevel.serializable)((ref tx) { ... });

// Server default — emits plain BEGIN, defers to postgresql.conf / ALTER ROLE / ALTER DATABASE
c.transaction!(OnSuccess.commit, IsolationLevel.serverDefault)((ref tx) { ... });

vibe.d (peque:vibe)

peque:vibe provides a fiber-aware wait strategy and connection pool for vibe.d applications. Instead of blocking the OS thread while waiting for PostgreSQL, control yields to the vibe.d event loop.

dependency "peque:vibe" version="~>0.1.0"

import peque;
import peque.vibe;

// Single connection with fiber-aware I/O
auto conn = Connection(params, VibeWaitStrategy());

// Connection pool — makeVibePool injects VibeWaitStrategy and non-blocking mode
auto pool = makeVibePool(8, [
    "dbname": "myapp",
    "user":   "app",
    "host":   "localhost",
    "port":   "5432",
]);

// Borrow a connection for the duration of a delegate; returned automatically
auto result = pool.borrow((ref Connection conn) {
    return conn.execParams("SELECT name FROM users WHERE id = $1", userId);
});

LISTEN / NOTIFY

peque exposes PostgreSQL's notification bus: listen()/unlisten() subscribe a connection to channels, and waitNotifications(Duration) delivers Notification { channel, payload, backendPid } values with a bounded wait — the shape a server-sent-events hub or cache invalidator needs.

import core.time: seconds;
import peque;

// The listening connection must be DEDICATED: autocommit (no open
// transactions — the server delivers notifications only between
// transactions), never pooled, owned by a single consumer loop.
auto conn = Connection(params);            // or Connection(params, VibeWaitStrategy())
conn.listen("events");                     // channel name is identifier-quoted

bool running = true;
while (running) {
    // Bounded wait doubles as the heartbeat tick: empty result on timeout.
    foreach (n; conn.waitNotifications(30.seconds))
        dispatch(n.channel, n.payload);
    // ...check stop flag, send SSE heartbeat, verify conn.status() here...
}

Publishing needs no dedicated API — pg_notify is a regular parameterized query, and NOTIFY is transactional (delivered on COMMIT, discarded on ROLLBACK):

conn.execParams("SELECT pg_notify($1, $2)", "events", payload);

getNotifications() is the non-blocking variant (drain only, never waits). waitNotifications drains libpq's buffer before waiting, so notifications that arrived during earlier traffic are returned immediately; a zero timeout makes it a pure non-blocking check.

Caveats worth knowing:

• Delivery happens only between transactions — keep the listening connection out of transactions and out of pools.
• Subscriptions do not survive reconnect. On a dead connection (conn.status() != CONNECTION_OK), build a fresh Connection and re-issue listen() for every channel (peque deliberately has no PQreset wrapper).
• The server deduplicates identical `(channel, payload)` notifications sent within one transaction.
• Payloads are limited to ~8000 bytes — send an ID, not a document.
• waitNotifications requires the connection's WaitStrategy to provide the optional timed overload bool wait(int fd, WaitMask mask, Duration timeout). PollWaitStrategy (default) and VibeWaitStrategy both do; a custom strategy without it keeps working for queries and getNotifications().

Running tests

Integration tests require a running PostgreSQL instance. Configure via environment variables (defaults shown):

POSTGRES_DB=peque-test \
POSTGRES_USER=peque \
POSTGRES_PASSWORD=peque \
POSTGRES_HOST=localhost \
POSTGRES_PORT=5432 \
dub test --config=unittestStatic

License

Mozilla Public License 2.0