ktm
/
v0.28.3

PostgreSQL

Talk to PostgreSQL from ktm with the ktm-pg package. This guide builds up a catalogue of books, one whole program at a time: connecting, running statements with parameters, reading rows into structs, NULLs, rows one by one, transactions, errors, a connection pool behind an HTTP API, batches and fast bulk loading, query timeouts, and LISTEN/NOTIFY.

What you need

docker run -d --name pg -p 5432:5432 -e POSTGRES_PASSWORD=secret postgres:17

Make a project and add the package:

mkdir books && cd books
ktm mod init example.com/me/books
ktm get github.com/ozzenerol/ktm-pg

Every step of this guide is a whole main.ktm for this project. Each one carries on from the tables the last one left, so run them in order. They connect with the server's address from the environment, so set it once in the terminal you build in:

export PGHOST=localhost PGUSER=postgres PGPASSWORD=secret PGDATABASE=postgres

Then build and run each step like this:

ktm build
./books

Connect and run statements

import "github.com/ozzenerol/ktm-pg";

i32 main() {
    pg.Conn *c = pg.connect();      /* PGHOST, PGUSER, PGPASSWORD, PGDATABASE */
    defer c.close();
    printf("connected to PostgreSQL %s\n", c.serverVersion());

    c.execScript(
        "CREATE TABLE IF NOT EXISTS books (\n"
        "    id     bigserial PRIMARY KEY,\n"
        "    title  text NOT NULL UNIQUE,\n"
        "    author text NOT NULL,\n"
        "    year   int\n"
        ");\n");

    i64 n = c.exec("INSERT INTO books (title, author, year) VALUES ($1, $2, $3) ON CONFLICT (title) DO NOTHING",
                   "Dune", "Frank Herbert", 1965);
    printf("%d row(s) inserted\n", n);

    i64 id = pg.queryOne<i64>(c, "INSERT INTO books (title, author, year) VALUES ($1, $2, $3) "
                                 "ON CONFLICT (title) DO UPDATE SET year = excluded.year RETURNING id",
                              "Neuromancer", "William Gibson", 1984);
    printf("Neuromancer has id %d\n", id);

    i64 count = pg.queryOne<i64>(c, "SELECT count(*) FROM books");
    printf("%d book(s)\n", count);
    return 0;
}
ktm build
./books
connected to PostgreSQL 17.11 (Debian 17.11-1.pgdg13+2)
1 row(s) inserted
Neuromancer has id 2
2 book(s)

Run it again and nothing is added: ON CONFLICT turns the inserts into no-ops.

./books
connected to PostgreSQL 17.11 (Debian 17.11-1.pgdg13+2)
0 row(s) inserted
Neuromancer has id 2
2 book(s)

Read rows into structs

import "github.com/ozzenerol/ktm-pg";
import reflect, vec;

struct Book {
    i64 id;
    String title;
    String author;
    i64 *year;      /* a pointer: null when the column is NULL */
};

void print(Book *b) {
    if (b.year) {
        printf("%3d  %s, %s (%d)\n", b.id, b.title.cstr(), b.author.cstr(), *b.year);
    } else {
        printf("%3d  %s, %s (year unknown)\n", b.id, b.title.cstr(), b.author.cstr());
    }
}

i32 main() {
    pg.Conn *c = pg.connect();
    defer c.close();

    i64 *noYear = null;
    c.exec("INSERT INTO books (title, author, year) VALUES ($1, $2, $3) ON CONFLICT (title) DO NOTHING",
           "Robert's Rules", "Henry M. Robert", noYear);
    c.exec("INSERT INTO books (title, author, year) VALUES ($1, $2, $3) ON CONFLICT (title) DO NOTHING",
           "The Left Hand of Darkness", "Ursula K. Le Guin", 1969);

    /* every row, into a vector of structs */
    vec.Vec<Book> all = pg.queryAll<Book>(c, "SELECT id, title, author, year FROM books ORDER BY id");
    defer pg.free(reflect.of(all));
    for (auto b : all) {
        print(&b);
    }

    /* one row */
    Book dune = pg.queryOne<Book>(c, "SELECT * FROM books WHERE title = $1", "Dune");
    defer pg.free(reflect.of(dune));
    printf("found: ");
    print(&dune);

    /* no row: queryOne throws pg.ERR_NO_ROWS */
    try {
        Book none = pg.queryOne<Book>(c, "SELECT * FROM books WHERE id = $1", 999);
        pg.free(reflect.of(none));
    } catch (Error e) {
        if (e.code == pg.ERR_NO_ROWS) {
            printf("no book 999\n");
        } else {
            throw(e);
        }
    }

    /* not a struct: the first column */
    vec.Vec<String> authors = pg.queryAll<String>(c, "SELECT DISTINCT author FROM books ORDER BY author");
    defer pg.free(reflect.of(authors));
    printf("%d authors, the first is %s\n", authors.len, authors.data[0].cstr());
    return 0;
}
ktm build
./books
  1  Dune, Frank Herbert (1965)
  2  Neuromancer, William Gibson (1984)
  3  Robert's Rules, Henry M. Robert (year unknown)
  4  The Left Hand of Darkness, Ursula K. Le Guin (1969)
found:   1  Dune, Frank Herbert (1965)
no book 999
4 authors, the first is Frank Herbert

Rows one by one

c.query returns the rows to read one at a time, without building structs. It suits a big result, or one you only print:

import "github.com/ozzenerol/ktm-pg";

i32 main() {
    pg.Conn *c = pg.connect();
    defer c.close();

    scoped pg.Rows r = c.query("SELECT id, title, year FROM books WHERE id > $1 ORDER BY id", 0);
    printf("%d columns: %s, %s, %s\n", r.cols, r.name(0), r.name(1), r.name(2));
    while (r.next()) {
        i64 id = r.int(0);
        u8[] title = r.text(1);       /* a view: valid until the next row */
        if (r.isNull(2)) {
            printf("%3d  %s\n", id, title);
        } else {
            printf("%3d  %s (%d)\n", id, title, r.int(2));
        }
    }
    return 0;
}
ktm build
./books
3 columns: id, title, year
  1  Dune (1965)
  2  Neuromancer (1984)
  3  Robert's Rules
  4  The Left Hand of Darkness (1969)

Transactions

import "github.com/ozzenerol/ktm-pg";

/* Moves `amount` from one account to another: both updates, or neither */
void transfer(pg.Conn *c, u8[] from, u8[] to, i64 amount) {
    pg.Tx tx = c.begin();
    defer tx.rollback();          /* does nothing once commit() ran */
    tx.exec("UPDATE accounts SET balance = balance - $1 WHERE name = $2", amount, from);
    tx.exec("UPDATE accounts SET balance = balance + $1 WHERE name = $2", amount, to);
    tx.commit();
}

void balances(pg.Conn *c) {
    scoped pg.Rows r = c.query("SELECT name, balance FROM accounts ORDER BY name");
    while (r.next()) {
        printf("  %s: %d\n", r.text(0), r.int(1));
    }
}

i32 main() {
    pg.Conn *c = pg.connect();
    defer c.close();

    c.execScript(
        "DROP TABLE IF EXISTS accounts;\n"
        "CREATE TABLE accounts (name text PRIMARY KEY, balance bigint NOT NULL CHECK (balance >= 0));\n"
        "INSERT INTO accounts VALUES ('ann', 100), ('bob', 20);\n");

    transfer(c, "ann", "bob", 30);
    printf("after ann -> bob 30:\n");
    balances(c);

    try {
        transfer(c, "bob", "ann", 500);
    } catch (Error e) {
        printf("bob -> ann 500 failed: %s\n", e.message);
    }
    printf("after the failed transfer:\n");
    balances(c);
    return 0;
}
ktm build
./books
after ann -> bob 30:
  ann: 70
  bob: 50
bob -> ann 500 failed: new row for relation "accounts" violates check constraint "accounts_balance_check" (SQLSTATE 23514); detail: Failing row contains (bob, -450).
after the failed transfer:
  ann: 70
  bob: 50

Errors

Every failure throws. For an error from the server, e.code is its SQLSTATE, and ktm-pg names the common ones:

import "github.com/ozzenerol/ktm-pg";

/* Adds a book; false when one with that title is already there */
bool addBook(pg.Conn *c, u8[] title, u8[] author) {
    try {
        c.exec("INSERT INTO books (title, author) VALUES ($1, $2)", title, author);
    } catch (Error e) {
        if (e.code == pg.UNIQUE_VIOLATION) {
            return false;
        }
        throw(e);
    }
    return true;
}

i32 main() {
    pg.Conn *c = pg.connect();
    defer c.close();

    if (!addBook(c, "Dune", "Frank Herbert")) {
        printf("Dune is already there\n");
    }

    try {
        c.exec("INSERT INTO books (title) VALUES ($1)", "No Author");
    } catch (Error e) {
        printf("%s\n", e.message);
    }
    try {
        c.exec("SELECT * FROM book");
    } catch (Error e) {
        printf("%s\n", e.message);
    }
    printf("the connection still works: %d books\n", pg.queryOne<i64>(c, "SELECT count(*) FROM books"));
    return 0;
}
ktm build
./books
Dune is already there
null value in column "author" of relation "books" violates not-null constraint (SQLSTATE 23502); detail: Failing row contains (6, No Author, null, null).
relation "book" does not exist (SQLSTATE 42P01)
the connection still works: 4 books

A pool behind an HTTP API

A server needs more than one connection: requests arrive together, on several threads. A pg.Pool keeps connections open and lends one to each query. This step turns the catalogue into a small JSON API with net.http:

import "github.com/ozzenerol/ktm-pg";
import json, net.http, reflect, str, vec;

@json struct Book {
    i64 id;
    String title;
    String author;
    i64 *year;
};

@json struct NewBook {
    String title;
    String author;
    i64 year;
};

pg.Pool *db;

/* GET /books */
void listBooks(http.Writer *w, http.Request *r) {
    vec.Vec<Book> books = pg.queryAll<Book>(db, "SELECT id, title, author, year FROM books ORDER BY id");
    defer pg.free(reflect.of(books));
    w.json(200, reflect.of(books));
}

/* GET /books/{id} */
void getBook(http.Writer *w, http.Request *r) {
    i64 id;
    if (!str.toInt(r.param("id"), &id)) {
        w.text(400, "bad id\n");
        return;
    }
    vec.Vec<Book> found = pg.queryAll<Book>(db, "SELECT id, title, author, year FROM books WHERE id = $1", id);
    defer pg.free(reflect.of(found));
    if (!found.len) {
        w.text(404, "no book %d\n", id);
        return;
    }
    w.json(200, reflect.of(found.data[0]));
}

/* POST /books with {"title": ..., "author": ..., "year": ...} */
void addBook(http.Writer *w, http.Request *r) {
    NewBook nb;
    try {
        nb = json.decode<NewBook>(r.body);
    } catch (Error e) {
        w.text(400, "%s\n", e.message);
        return;
    }
    defer json.free(reflect.of(nb));
    try {
        i64 id = pg.queryOne<i64>(db, "INSERT INTO books (title, author, year) VALUES ($1, $2, $3) RETURNING id",
                                  nb.title, nb.author, nb.year);
        w.text(201, "%d\n", id);
    } catch (Error e) {
        if (e.code == pg.UNIQUE_VIOLATION) {
            w.text(409, "there is already a book called %s\n", nb.title.cstr());
            return;
        }
        throw(e);
    }
}

i32 main() {
    db = pg.open();               /* connects when a query first needs it */
    defer db.close();
    db.ping();                    /* a wrong address or password shows here, at start-up */

    http.Mux mux;
    mux.handle("GET /books", &listBooks);
    mux.handle("GET /books/{id}", &getBook);
    mux.handle("POST /books", &addBook);
    printf("listening on :8080\n");
    return http.listen(8080, &mux);
}

Start it:

ktm build
./books

Then, from a second terminal:

curl localhost:8080/books
[{"id":1,"title":"Dune","author":"Frank Herbert","year":1965},{"id":2,"title":"Neuromancer","author":"William Gibson","year":1984},{"id":3,"title":"Robert's Rules","author":"Henry M. Robert","year":null},{"id":4,"title":"The Left Hand of Darkness","author":"Ursula K. Le Guin","year":1969}]
curl localhost:8080/books/1
curl localhost:8080/books/999
curl localhost:8080/books/x
{"id":1,"title":"Dune","author":"Frank Herbert","year":1965}
no book 999
bad id
curl -X POST localhost:8080/books -d '{"title": "Hyperion", "author": "Dan Simmons", "year": 1989}'
curl -X POST localhost:8080/books -d '{"title": "Hyperion", "author": "Dan Simmons", "year": 1989}'
curl -X POST localhost:8080/books -d '{"title": 5}'
7
there is already a book called Hyperion
1:11: title: expected a string, got a number

Stop the server with Ctrl+C.

Batches and bulk loading

Each exec waits for its answer before the next one goes out. A batch sends several statements at once, and binary COPY loads many rows faster than any INSERT:

import "github.com/ozzenerol/ktm-pg";
import date;

struct Reading {
    i64 sensor;
    f64 celsius;
};

i32 main() {
    pg.Conn *c = pg.connect();
    defer c.close();
    c.execScript(
        "DROP TABLE IF EXISTS readings;\n"
        "CREATE TABLE readings (sensor bigint NOT NULL, celsius float8 NOT NULL);\n");

    /* a batch: several statements in one round trip, all or nothing */
    pg.Batch b;
    defer b.free();
    b.add("INSERT INTO readings VALUES ($1, $2)", 1, 20.5);
    b.add("INSERT INTO readings VALUES ($1, $2)", 2, 21.0);
    b.add("UPDATE readings SET celsius = celsius + $1 WHERE sensor = $2", 0.25, 1);
    i64[] counts = c.execBatch(&b);
    printf("batch: %d statements, the last changed %d row(s)\n", counts.len, counts[counts.len - 1]);
    counts.free();

    /* binary COPY: the fastest way in for many rows */
    i64 t0 = date.monoMs();
    pg.CopyInto<Reading> in = c.copyInto<Reading>("readings");
    for (i64 i = 0; i < 100000; i++) {
        Reading r;
        r.sensor = i % 50;
        r.celsius = 15.0 + (i % 100) / 10.0;
        in.put(r);
    }
    i64 n = in.end();
    printf("copied %d rows in %d ms\n", n, date.monoMs() - t0);

    f64 avg = pg.queryOne<f64>(c, "SELECT avg(celsius) FROM readings");
    printf("%d rows, %.2f degrees on average\n", pg.queryOne<i64>(c, "SELECT count(*) FROM readings"), avg);
    return 0;
}
ktm build
./books
batch: 3 statements, the last changed 1 row(s)
copied 100000 rows in 23 ms
100002 rows, 19.95 degrees on average

Query timeouts

import "github.com/ozzenerol/ktm-pg";

i32 main() {
    pg.Conn *c = pg.connect();
    defer c.close();

    c.setQueryTimeout(500);       /* ms, for every query on c from now on */
    try {
        c.exec("SELECT pg_sleep(3)");
    } catch (Error e) {
        if (e.code == pg.QUERY_CANCELED) {
            printf("cancelled: %s\n", e.message);
        } else {
            throw(e);
        }
    }
    printf("the connection still works: %d\n", pg.queryOne<i64>(c, "SELECT 1 + 1"));
    return 0;
}
ktm build
./books
cancelled: timeout after 500 ms: canceling statement due to user request (SQLSTATE 57014)
the connection still works: 2

LISTEN and NOTIFY

Postgres can tell a program that something happened, without the program asking over and over:

import "github.com/ozzenerol/ktm-pg";

i32 main() {
    pg.Conn *listener = pg.connect();     /* a connection of its own for listening */
    defer listener.close();
    pg.Conn *c = pg.connect();
    defer c.close();

    listener.listen("new_books");

    /* another program, or a trigger, would do this */
    c.notify("new_books", "Hyperion");
    c.exec("SELECT pg_notify('new_books', $1)", "The Fall of Hyperion");

    pg.Notification n;
    while (listener.waitNotification(&n, 1000)) {     /* false after 1 s with none */
        printf("%s: %s\n", n.channel, n.payload);
        n.free();
    }
    printf("no more for 1 s\n");
    return 0;
}
ktm build
./books
new_books: Hyperion
new_books: The Fall of Hyperion
no more for 1 s

Connecting securely

ktm-pg uses TLS when the server offers it (sslmode=prefer, as libpq does). For a server across a network, require it and check the server's certificate:

pg.Conn *c = pg.connect("postgres://app@db.example.com/app?sslmode=verify-full&sslrootcert=/etc/ssl/db-ca.pem");
sslmodeWhat it does
disablenever TLS
preferTLS when the server has it (the default)
requireTLS, without checking the certificate
verify-caTLS, and the certificate must come from a CA in sslrootcert
verify-fullas verify-ca, and the certificate must name the host

Passwords are sent with SCRAM-SHA-256, the Postgres default, and with TLS also bound to the connection (SCRAM-SHA-256-PLUS).

Reference

CallWhat it does
pg.connect(url)a pg.Conn *; the URL, else the PG* environment
pg.open(url, &opts)a pg.Pool *, safe from every thread
c.close(), db.close()ends a connection or a pool
c.exec(sql, args...)one statement; the rows it changed
c.execScript(sql)several statements, no parameters
c.query(sql, args...)pg.Rows to read with next()
pg.queryAll<T>(db, sql, args...)every row, as a vec.Vec<T>
pg.queryOne<T>(db, sql, args...)the first row as a T; pg.ERR_NO_ROWS when none
pg.free(reflect.of(x))frees what scanning allocated
c.begin()a pg.Tx: commit(), rollback()
c.execBatch(&b)a pg.Batch in one round trip
c.copyInto<T>(table)binary COPY from structs: put, end, abort
c.setQueryTimeout(ms)a time limit for every query
c.listen(ch), c.waitNotification(&n, ms), c.notify(ch, payload)LISTEN and NOTIFY
db.acquire(), c.release()a pool's connection to yourself

db in the table is a pg.Conn *, a pg.Tx * or a pg.Pool *.

Rules of thumb