postgres#

A PostgreSQL client with an API close to node-postgres. The driver is the package's native library (Rust over tokio-postgres and rustls, on Velt's own sockets). Every call that talks to the server is async.

import { connect, createPool, Client, Pool, PoolOptions, Transaction, CopyReader, CopyWriter, PgError } from "postgres";

Adding it to a program#

There is no public package registry yet. Depend on the package in the Velt repository, from git or from a local checkout, in package.vlt:

import type { Package } from "velt/package";

export const pkg: Package = {
  name: "app",
  version: "0.1.0",
  dependencies: {
    postgres: { git: "https://github.com/velt-lang/velt", branch: "main", path: "packages/postgres" },
    // Or a local checkout of the repository:
    // postgres: { path: "../velt/packages/postgres" },
  },
};

The driver is a Rust crate (the package's native/ directory), so both forms build it with cargo the first time and need a Rust toolchain. A path dependency builds it directly; a git dependency builds it only when you allow that with VELT_NATIVE_FROM_SOURCE=1, since the build runs the package's build scripts. A package published to a registry carries prebuilt libraries, and then needs no Rust (packages with native code).

A first query#

import { connect, createPool } from "postgres";

struct User {
  id: i64;
  name: string;
}

const db = await connect("postgres://app@localhost/app");
await db.batch("CREATE TEMP TABLE users (id BIGSERIAL PRIMARY KEY, name TEXT UNIQUE)");
await db.execute("INSERT INTO users (name) VALUES (:name)", { name: "Ann" });
const users: User[] = await db.query("SELECT id, name FROM users WHERE id > $1", [0]);
const ann: User | null = await db.queryOne("SELECT * FROM users WHERE name = :n", { n: "Ann" });
console.log(users.length, ann?.id); // 1 1
db.close();

const pool = createPool({ url: "postgres://app@localhost/app", max: 4 });
console.log(await pool.run("SELECT 1")); // 1
pool.end();

Connecting#

connect(url): Promise<Client>. url is postgres://user:password@host:port/db?options or a libpq string (host=… user=… dbname=…). Supported options include connect_timeout, application_name and sslmode (TLS).

A Client is one connection, and a handle: pass it around and capture it in tasks. Concurrent queries on one client are pipelined on its connection. close() releases it; after close() through one copy, every copy throws ECLOSED (closing again does nothing).

Queries#

Passing {} as params also works for a statement without parameters. batch(sql) runs a script of ;-separated statements (no parameters, not prepared).

Prepared statements are cached per connection (256 per connection, the least recently used evicted), so repeating a query costs one round trip.

Parameters#

Parameters are a value bound as its JSON.stringify text would be (velt/db's bindParams).

• An array binds $1, $2, … by position.
• An object binds :name or $name by field name. They are rewritten to $n once per statement; ::type casts, strings, quoted identifiers and comments are left alone.
• Values are converted for the parameter types the server inferred: integers are range-checked for int2/int4, numbers also go to numeric and text, bools to bool, u8[] to bytea, and a string to text and json/jsonb (pass JSON.stringify(doc)). A string for any other type (uuid, date, timestamptz, numeric, arrays such as "{1,2}", enums) is parsed by the server. Array elements share one type, so mixed positional values are either an object or all strings.

Rows into typed values#

Rows are decoded by velt/db's decodeRows, as JSON.parse<T[]> would decode them: a column fills the field with the same name (rename with AS).

Other types (ranges, composites, geometric types, …) throw ENOTSUP: cast them, for example d::text. A row that doesn't fit T throws EMISMATCH. Fractions of a second use 3 digits for whole milliseconds, 6 otherwise, and are left out when zero.

Batches#

batchQuery<T, P>(sql, paramSets: P[]): Promise<T[][]>, batchQueryOne<T, P>(sql, paramSets): Promise<(T | null)[]> and batchExecute<P>(sql, paramSets): Promise<i64[]> run one statement with many parameter sets in one round trip (one message group with a single Sync). The server runs them as one implicit transaction: if one execution fails, the whole batch throws that error and, outside a transaction, none of its writes are kept.

const counts = await db.batchExecute("INSERT INTO users (name) VALUES ($1)", [["Bob"], ["Cy"]]);
console.log(counts); // [1, 1]

Transactions#

transaction(fn) (fn: async (tx: Client) => T, which may throw) commits after fn returns and returns its result. If fn throws, or any operation on the client failed in the meantime (even one fn caught), it rolls back and throws that error (the first failed operation's). Nested calls use savepoints: a failed inner transaction undoes only its own work.

try {
  await db.transaction(async (tx) => {
    await tx.execute("INSERT INTO users (name) VALUES ($1)", ["Bob"]);
    try {
      await tx.execute("INSERT INTO users (name) VALUES ($1)", ["Ann"]);
    } catch (e) {
      console.log(e.code, e.constraint); // 23505 users_name_key
    }
    return 0;
  });
} catch (e) {
  console.log("rolled back:", e.code); // Bob was not inserted either
}

await db.begin() returns a Transaction with async commit() and rollback() for explicit control; inTransaction says whether one is open.

Pools#

createPool({ url, max? }): Pool (PoolOptions; max defaults to 10). A bad connection string throws here; connections open on first use. A Pool has the same query, queryOne, execute, select, selectOne, run, batch, batchQuery, batchQueryOne and batchExecute methods, each running on a free connection (waiting when all max are busy), plus:

• connect(): Promise<Client>: a dedicated connection; its close() returns it to the pool. A pooled connection still in a transaction when its client is closed is dropped, not reused.
• transaction(fn) on a dedicated connection.
• idleCount and end().

Pools are safe to use from many spawned tasks, and a Pool is a handle like a Client: a request handler can capture one.

COPY#

COPY is streamed both ways:

• copyFrom(sql): Promise<CopyWriter> starts COPY … FROM STDIN (for example "COPY items (id, name) FROM STDIN (FORMAT csv)"). await w.write(text) and writeBytes(u8[]) send chunks split anywhere, await w.end() completes it and returns the rows copied, w.abort() cancels it (nothing is copied).
• copyTo(sql): Promise<CopyReader> starts COPY … TO STDOUT. await r.read() returns the data received so far (whole rows in the text and CSV formats, up to about 64 KiB per call), readBytes() the same as bytes (binary format), both null at the end; r.close() stops early. A CopyReader is an AsyncIterable<string, PgError>: for await (const chunk of r) reads the text chunks to the end (leaving the loop early keeps the copy open: close() it).

const w = await db.copyFrom("COPY users (name) FROM STDIN (FORMAT csv)");
await w.write("Dee\nEd\n");
console.log(await w.end()); // 2

const r = await db.copyTo("COPY users TO STDOUT (FORMAT csv)");
for await (const chunk of r) {
  console.log(chunk);
}

The connection is busy until the copy ends; a failed copy counts against an enclosing transaction.

Errors#

PgError { code, message, detail, constraint }. code is the server's SQLSTATE (for example "23505" unique violation, "42P01" undefined table, "42601" syntax error, "25P02" statement in an aborted transaction). Failures outside the server use names: "ECONNREFUSED" (and the other velt/io names), "ECONNRESET" (connection lost), "ECLOSED" (closed client or ended pool), "EINVAL" (connection string or parameters), "ETLS", "ENOTSUP" (a column type that can't be decoded) and "EMISMATCH" (a row that doesn't fit T).

TLS#

TLS is rustls, chosen with sslmode:

Trusted roots are Mozilla's plus the PEM file named by sslrootcert, which also upgrades require to verify-ca, as in libpq. SCRAM channel binding isn't offered.

Testing it#

The package's tests (velt test in packages/postgres) need a PostgreSQL server: set VELT_TEST_PG_URL (postgres://user:password@host/db). Without it they do nothing. Tests of your own program that use the database need a server the same way; pass its URL in an environment variable.