moonpg

A pure MoonBit PostgreSQL client, wire protocol from scratch.

postgres
postgresql
wire-protocol
database
moon add jaredzhou/moonpg@0.8.11
Download zip
Author
Version
0.8.11
License
Apache-2.0
Last updated
21 days ago
Downloads
69
README

#jaredzhou/moonpg

A pure MoonBit PostgreSQL client — wire protocol from scratch, zero C dependencies. Feature-complete with a simple, ergonomic API.

#Features

  • Query & Fetchquery/query_one/execute for manual control, fetch/fetch_one with FromRow for typed auto-mapping to structs or tuples.
  • Type-safe codecToValue encodes parameters, FromValue decodes results; all base types, Json, Timestamp, Decimal, UUID, Option<T> supported.
  • Connection pool — bounded pool with acquire/release, min-idle, idle-timeout, health check, background maintenance.
  • Transactionsbegin_tx / commit / rollback + begin_func auto-commit/rollback.
  • Async — multiple connections run concurrently; slow queries never block others.
  • COPY — streaming bulk insert from iterators.
  • LISTEN / NOTIFY — async notification support.
  • TLSsslmode support (disable, require, verify-ca, verify-full) with client certificates.

#Install

moon add jaredzhou/moonpg

#Quickstart

let conn = @moonpg.connect("postgres://user:pw@localhost:5432/db")

// Execute DDL / DML
conn.execute("CREATE TABLE users (id SERIAL PRIMARY KEY, name TEXT, email TEXT)") |> ignore
conn.execute(
"INSERT INTO users (name, email) VALUES ($1, $2)",
params=["alice", "a@b.com"],
) |> ignore

// fetch — typed array of rows (auto-closes)
let names : Array[String] = &QueryExecutor::fetch(conn, "SELECT name FROM users ORDER BY id")
let count : Int = &QueryExecutor::fetch_one(conn, "SELECT COUNT(*) FROM users")

// fetch with tuples
let users : Array[(Int, String)] = &QueryExecutor::fetch(conn, "SELECT id, name FROM users")
let (id, name) = &QueryExecutor::fetch_one(conn, "SELECT id, name FROM users WHERE id = $1", params=[1])

// query — manual iteration (MUST close rows in try-catch)
let rows = conn.query("SELECT id, name FROM users")
try {
while rows.has_next() {
let row = rows.get_row()
let id : Int = row.get(0)
let name : String = row.get_by_name("name")
}
rows.close()
} catch {
e => { rows.close(); raise e }
}

#Query

#fetch / fetch_one

fetch and fetch_one collect rows into typed results and auto-close — no manual rows.close() or try-catch needed. They're the recommended way to read data.

let conn = @moonpg.connect(conninfo)

// Scalar
let count : Int = &QueryExecutor::fetch_one(conn, "SELECT COUNT(*) FROM users")

// Nullable
let email : String? = &QueryExecutor::fetch_one(conn, "SELECT email FROM users WHERE id = $1", params=[1])

// Tuples — no struct needed
let pairs : Array[(Int, String)] = &QueryExecutor::fetch(conn, "SELECT id, name FROM users")
let (id, name, email) : (Int, String, String) = &QueryExecutor::fetch_one(conn,
"SELECT id, name, email FROM users WHERE id = $1", params=[1],
)

// Custom FromRow struct
impl FromRow for User with fn from_row(r : Row) -> User raise PgError {
User::{ id: r.get(0), name: r.get(1), email: r.get(2) }
}
let users : Array[User] = &QueryExecutor::fetch(conn, "SELECT id, name, email FROM users")
// Or cast to &QueryExecutor for dot-syntax:
let q : &QueryExecutor = conn
let count : Int = q.fetch_one("SELECT COUNT(*) FROM users")
let names : Array[String] = q.fetch("SELECT name FROM users")

#query / query_one

query returns a &Rows iterator. You must call rows.close() to drain results and (for pool connections) return the connection. Always use try-catch so close() runs even on decode errors.

// query — iterate rows manually
let rows = conn.query("SELECT id, name FROM users")
try {
while rows.has_next() {
let row = rows.get_row()
let id : Int = row.get(0)
let name : String = row.get_by_name("name")
println("\{id}: \{name}")
}
rows.close()
} catch {
e => { rows.close(); raise e }
}

// query_one — single row
let row = conn.query_one("SELECT id FROM users WHERE name = $1", params=["alice"])
let id : Int = row.get(0)

#execute

conn.execute(
"UPDATE users SET email = $1 WHERE id = $2",
params=[null, 42], // null → SQL NULL
) |> ignore

#Pool

let pool = Pool::new(PoolConfig::new(
"postgres://user:pw@localhost:5432/db",
max_conns=10,
min_idle=2,
))

// Auto-acquire + auto-release — fetch/rows.close() returns conn to pool
let names : Array[String] = &QueryExecutor::fetch(pool, "SELECT name FROM users")
pool.execute("INSERT INTO users (name) VALUES ($1)", params=["bob"]) |> ignore

// Manual query with try-catch
let rows = pool.query("SELECT id FROM users")
try {
while rows.has_next() { ... }
rows.close()
} catch {
e => { rows.close(); raise e }
}

// Explicit acquire
let pc = pool.acquire()
pc.execute("DELETE FROM users WHERE id = $1", params=[1]) |> ignore
pc.release()

// Inspect pool
let stats = pool.stats()
println("active=\{stats.active_connections} idle=\{stats.idle_connections}")

// Health check + background maintenance
let pool2 = Pool::new(PoolConfig::new(
conninfo,
max_conns=10,
min_idle=2,
max_idle_sec=300,
max_lifetime_sec=3600,
health_check=true,
maintenance_interval_sec=60,
))
pool2.start_maintenance()

#Transactions

let conn = @moonpg.connect(conninfo)

// begin_func — auto-commit on success, auto-rollback on error
let result = begin_func(conn, async fn(tx) {
tx.execute("INSERT INTO users (name) VALUES ($1)", params=["alice"]) |> ignore
&QueryExecutor::fetch_one(tx, "SELECT id FROM users WHERE name = $1", params=["alice"])
})

// Manual transaction — try { commit } catch { rollback }
let tx = conn.begin_tx()
try {
tx.execute("UPDATE users SET name = $1 WHERE id = $2", params=["bob", 1]) |> ignore
let name : String = &QueryExecutor::fetch_one(tx, "SELECT name FROM users WHERE id = $1", params=[1])
tx.commit()
} catch {
e => { tx.rollback(); raise e }
}

// Pooled transaction — connection auto-returns to pool on commit/rollback
let pool = Pool::new(PoolConfig::new(conninfo, max_conns=4))
let tx2 = pool.begin_tx()
try {
tx2.execute("DELETE FROM users WHERE id = $1", params=[99]) |> ignore
tx2.commit()
} catch {
e => { tx2.rollback(); raise e }
}

#ToValue / FromValue

// ToValue — MoonBit → PostgreSQL
// Built-in impls: Int, Int64, Double, Bool, String, Bytes, Json,
// Timestamp, Decimal, UUID, Option<T>, Array<T>
let params = [42, 3.14, true, "hello", null] // null → SQL NULL
conn.execute("INSERT INTO t (a, b, c, d, e) VALUES ($1, $2, $3, $4, $5)", params=params)

// FromValue — PostgreSQL → MoonBit
let row = conn.query_one("SELECT a, b, c, d, e FROM t")
let a : Int = row.get(0) // strict: raises on NULL
let b : String? = row.get(1) // nullable: NULL → None
let c : Bool = row.get_by_name("c") // by column name
let d : Json = row.get(3) // jsonb → Json
let e : Timestamp = row.get(4) // timestamptz → Unix µs

// Arrays — PostgreSQL array columns
conn.execute(
"INSERT INTO items (tags) VALUES ($1)",
params=[["red", "green", "blue"]],
) |> ignore
let row2 = conn.query_one("SELECT tags FROM items")
let tags : Array[String] = row2.get(0) // strict: no NULL elements
let tags2 : Array[String?] = row2.get(0) // nullable: NULL elements → None

// Custom impl
impl ToValue for MyType with fn to_value(self) -> Value {
Value::String(self.to_json())
}
impl FromValue for MyType with fn from_value(v : Value) -> MyType raise ValueError {
match v { Value::String(s) => MyType::from_json(s); _ => raise ... }
}

#FromRow

// Single-column rows: built-in impls for all FromValue types
let count : Int = &QueryExecutor::fetch_one(conn, "SELECT COUNT(*) FROM users")
let name : String? = &QueryExecutor::fetch_one(conn, "SELECT name FROM users WHERE id = $1", params=[1])

// Custom struct
impl FromRow for User with fn from_row(r : Row) -> User raise PgError {
User::{ id: r.get(0), name: r.get(1), email: r.get(2) }
}
let users : Array[User] = &QueryExecutor::fetch(conn, "SELECT id, name, email FROM users")

// Tuple impls (2–10)
let pairs : Array[(Int, String)] = &QueryExecutor::fetch(conn, "SELECT id, name FROM users")
let (id, name, email) : (Int, String, String) = &QueryExecutor::fetch_one(conn,
"SELECT id, name, email FROM users WHERE id = $1", params=[1],
)

#Connection properties

let pid = conn.backend_pid() // server process ID
let ver = conn.param("server_version") // e.g. Some("16.4")
let tz = conn.param("TimeZone") // e.g. Some("UTC")
if conn.is_closed() { ... }

#TLS / SSL

// System CA
@moonpg.connect("postgres://user:pw@host/db?sslmode=require")

// Custom CA
@moonpg.connect("postgres://user:pw@host/db?sslmode=verify-ca&sslrootcert=/etc/ca.pem")

// Client certificate
@moonpg.connect(
"postgres://user:pw@host/db?sslmode=require&sslcert=/etc/certs/client.pem&sslkey=/etc/certs/client.key",
)

Supported sslmodes: disable, allow, prefer, require, verify-ca, verify-full.

#LISTEN / NOTIFY

@async.with_task_group() group => {
let listener = conn.listen("events", group)

group.spawn_bg() () => {
let c2 = @moonpg.connect(conninfo)
c2.notify("events", payload="hello")
}

let notif = listener.recv()
println("\{notif.channel}: \{notif.payload}")
}

#COPY protocol

// Bulk insert from an iterator — one row in memory at a time
conn.copy_in("COPY users (name, age) FROM STDIN", ["alice\t30\n", "bob\t25\n"].iter())

// Streaming COPY writer
let w = conn.begin_copy("users", ["name", "age"])
w.write_row(["alice", 30])
w.write_row(["bob", 25])
let result = w.finish()

#Connection timeouts

// TCP connect timeout (seconds)
@moonpg.connect("postgres://host/db?connect_timeout=5")

// Server-side statement timeout (milliseconds)
@moonpg.connect("postgres://host/db?statement_timeout=30000")

#Target session attributes

@moonpg.connect("postgres://host/db?target_session_attrs=read-write")

Values: any (default), read-write, read-only, primary, standby, prefer-standby.

#Architecture

See arch.md for a detailed walkthrough of the codebase.

#Run the tests

# Default connection moon test --target native # Custom connection PGCONN="postgres://user:pass@localhost:5432/mydb" moon test --target native

#Auth tests (optional)

env varexample connstrauth method
PG_PLAIN_CONNpostgres://moonpg_plain:plain_pass@localhost:5432/moonpg_testpassword
PG_MD5_CONNpostgres://moonpg_md5:md5_pass@localhost:5432/moonpg_testmd5
PG_SCRAM_CONNpostgres://moonpg_scram:scram_pass@localhost:5432/moonpg_testscram-sha-256

#License

Apache-2.0

#
Format

Data format for a PostgreSQL cell or parameter. Re-exported from @value.

#
Value

PostgreSQL value type. Re-exported from @value.

#
Closer

pub(open) trait Closer {
fn close(Self) -> Unit
}

Types that can be closed to release resources.

#
FromRow

pub(open) trait FromRow {
fn from_row(Row) -> Self raise PgError
}

Types that can be constructed from a single database Row.

Implement this trait for custom types to enable use with fetch / fetch_one.

impl FromRow for User with fn from_row(r : Row) -> User raise PgError {
User::{ id: r.get(0), name: r.get(1), email: r.get(2) }
}

let users : Array[User] = pool.fetch("SELECT id, name, email FROM users")
impl FromRow for Bool
impl FromRow for Int
impl FromRow for Int64
impl FromRow for Double
impl FromRow for String
impl FromRow for Option[T]
impl FromRow for Bytes
impl FromRow for Json
impl FromRow for Decimal
impl FromRow for UUID
impl FromRow for Tuple2[A, B]
impl FromRow for Tuple3[A, B, C]
impl FromRow for Tuple4[A, B, C, D]
impl FromRow for Tuple5[A, B, C, D, E]
impl FromRow for Tuple6[A, B, C, D, E, F]
impl FromRow for Tuple7[A, B, C, D, E, F, G]
impl FromRow for Tuple8[A, B, C, D, E, F, G, H]
impl FromRow for Tuple9[A, B, C, D, E, F, G, H, I]
impl FromRow for Tuple10[A, B, C, D, E, F, G, H, I, J]

#
FromValue

pub(open) trait FromValue {
fn from_value(
Value
) -> Self raise ValueError
}

Types that can be decoded from a Value.

Implementations should be strict: raise ValueError when the Value variant doesn't match the expected type.

impl FromValue for MyType with fn from_value(v : Value) -> MyType raise ValueError {
match v {
Value::String(s) => parse_my_type(s)
_ => raise ValueError::ValueError("expected Value::String, got ...")
}
}
impl FromValue for Bool
impl FromValue for Int
impl FromValue for Int64
impl FromValue for Double
impl FromValue for String
impl FromValue for Option[T]
impl FromValue for Bytes
impl FromValue for Array[T]

#
QueryExecutor

pub(open) trait QueryExecutor {
async fn query(Self, String, params? : Array[&ToValue]) -> &Rows raise PgError
async fn query_one(Self, String, params? : Array[&ToValue]) -> Row raise PgError
async fn execute(Self, String, params? : Array[&ToValue]) -> ExecResult raise PgError
}

Types that can execute SQL queries.

#
Rows

pub(open) trait Rows {
async fn has_next(Self) -> Bool raise PgError
fn get_row(Self) -> Row raise PgError
fn columns(Self) -> Array[
FieldDescription
]
async fn close(Self) -> Unit raise PgError
}

A pull-based row reader. Both ConnRows (bare connection) and PoolRows (pooled connection) implement this trait.

while rows.has_next() {
let row = rows.get_row()
let v : Int = row.get(0)
}
rows.close()

#
ToValue

pub(open) trait ToValue :
Debug
{
fn to_value(Self) ->
Value

}

Types that can be converted to a Value for use as SQL parameters.

impl ToValue for MyType with fn to_value(self) -> Value {
...
}
impl ToValue for Bool
impl ToValue for Int
impl ToValue for Int64
impl ToValue for Double
impl ToValue for String
impl ToValue for Bytes
impl ToValue for Value
pub(open) trait Tx : QueryExecutor {
async fn commit(Self) -> Unit raise PgError
async fn rollback(Self) -> Unit raise PgError
}

A database transaction. Extends QueryExecutor with commit/rollback. Concrete impl: DbTx. Users can impl this trait on mock types for testing.

#
TxBeginner

pub(open) trait TxBeginner {
async fn begin_tx(Self, opts? : TxOptions) -> &Tx raise PgError
}

Types that can start a transaction.

#
PgError

pub(all) suberror PgError {
ConnectionError(String)
QueryError(String)
NoRows
} derive(
Debug
)

#
ValueError

pub(all) suberror ValueError {
ValueError(String)
} derive(
Debug
)

Value conversion error.

#
ConnRows

pub(all) struct ConnRows {
reader :
ResultReader

}

Rows from a bare Connection. Wraps a wire ResultReader.
impl Rows for ConnRows

#
ConnStatus

pub enum ConnStatus {
OK
Bad
} derive(Eq,
Debug
)

Connection status.

#
Connection

pub struct Connection {
conn :
RawConn

}

Wraps a wire-protocol RawConn.

#
Connection::backend_pid

fn Connection::backend_pid(self : Connection) -> Int

Return the backend process ID assigned by PostgreSQL.

#
Connection::begin_copy

async fn Connection::begin_copy(self : Connection, table : String, columns : Array[String]) -> CopyWriter raise PgError

Begin a COPY ... FROM STDIN operation on the given table and columns.

Generates COPY table ("col1", "col2") FROM STDIN, sends the query, and returns a CopyWriter that accepts rows one at a time.

#
Connection::copy_in

async fn Connection::copy_in(self : Connection, sql : String, rows : Iter[String]) -> ExecResult raise PgError

Execute COPY ... FROM STDIN with rows from an iterator.

Each element is one line of COPY data (tab-separated, newline-terminated). Rows are pulled one at a time — only a single row is held in memory.

conn.copy_in("COPY t FROM STDIN", my_rows.iter())

#
Connection::deallocate

async fn Connection::deallocate(self : Connection, name : String) -> Unit raise PgError

Deallocate a prepared statement.

Sends Close(S) + Flush and waits for CloseComplete. Returns silently on success — raises PgError if the statement doesn't exist or the connection is broken.

Passing an empty string closes the unnamed statement.

#
Connection::is_closed

fn Connection::is_closed(self : Connection) -> Bool

Return true if the connection has been closed.

#
Connection::listen

async fn[X] Connection::listen(self : Connection, channel : String, group :
TaskGroup
[X]) -> Listener raise PgError

Start listening for notifications on the given channel.

Executes LISTEN channel, spawns a background receive loop on group, and returns a Listener. The connection is dedicated to listening.

@async.with_task_group() group => {
let listener = conn.listen("events", group)
for ;; {
let notif = listener.recv()
group.spawn_bg() () => { handle(notif) }
}
}

#
Connection::notify

async fn Connection::notify(self : Connection, channel : String, payload? : String) -> Unit raise PgError

Send a notification on the given channel.

An optional payload string can be included.

#
Connection::param

fn Connection::param(self : Connection, key : String) -> String?

Return a server parameter value (e.g. "server_version", "TimeZone").

Returns None if the parameter was not reported by the server during startup.

#
Connection::server_version

fn Connection::server_version(self : Connection) -> Int

Return the server version as an integer (e.g. 180004 for 18.0.4).

#
Connection::status

fn Connection::status(self : Connection) -> ConnStatus

Return the current connection status.

#
Connection::unlisten

async fn Connection::unlisten(self : Connection, channel : String) -> Unit raise PgError

Stop listening for notifications on the given channel.

Passing "*" unlistens from all channels.

#
CopyWriter

pub(all) struct CopyWriter {
conn :
RawConn

}

Streaming writer for COPY ... FROM STDIN.

Created via Connection::begin_copy. Rows are encoded and sent one at a time — only a single row is held in memory.

Example

let w = conn.begin_copy("users", ["name", "age"]) w.write_row(["Alice", 30]) w.write_row(["Bob", 25]) let result = w.finish()

#
CopyWriter::finish

async fn CopyWriter::finish(self : CopyWriter) -> ExecResult raise PgError

Finish the COPY operation.

Sends CopyDone, reads the server completion, and returns the ExecResult.

#
CopyWriter::write_row

async fn CopyWriter::write_row(self : CopyWriter, row : Array[&ToValue]) -> Unit raise PgError

Write a single row to the COPY stream.

Each value is encoded to COPY text format (tab-separated, \N for NULL, backslash-escaped strings) and sent immediately — no batching in memory.

#
DbTx

pub(all) struct DbTx {
conn : Connection
}

Concrete transaction wrapping a Connection.
impl Tx for DbTx

#
ExecResult

pub struct ExecResult {
tag :
CommandTag

}

Execution result (for INSERT/UPDATE/DELETE/DDL).

#
ExecResult::affected_rows

fn ExecResult::affected_rows(self : ExecResult) -> Int

Return the number of rows affected by an INSERT/UPDATE/DELETE.

#
ExecResult::close

fn ExecResult::close(_self : ExecResult) -> Unit

Close the execution result. No-op.

#
IdleConn

pub(all) struct IdleConn {
conn : Connection
created_at : Int64
idle_since : Int64
}

#
IsolationLevel

pub(all) enum IsolationLevel {
ReadCommitted
RepeatableRead
Serializable
}

Transaction isolation level.

#
Listener

Receiver for PostgreSQL notifications.

Created by Connection::listen. Call recv() to block until a notification arrives.

#
Listener::recv

Block until a notification arrives.

#
Pool

pub(all) struct Pool {
queue :
Queue
[IdleConn]
conninfo : String
max_conns : Int
min_idle : Int
max_idle_sec : Int
max_lifetime_sec : Int
health_check : Bool
maintenance_interval_sec : Int
count :
Ref
[Int]
idle_count :
Ref
[Int]
running :
Ref
[Bool]
acquire_count :
Ref
[Int64]
acquire_wait_count :
Ref
[Int64]
acquire_wait_duration :
Ref
[Int64]
}

impl Closer for Pool
impl TxBeginner for Pool

#
Pool::acquire

async fn Pool::acquire(self : Pool) -> PoolConn raise PgError

#
Pool::idle

fn Pool::idle(self : Pool) -> Int

Number of idle connections currently in the pool.

#
Pool::maintain

async fn Pool::maintain(self : Pool) -> Unit

Run one round of maintenance:

  1. Drain all idle connections from the queue, closing expired ones.
  2. Put the good ones back.
  3. Create new connections if idle count is below min_idle.

Called automatically by the background maintenance loop; may also be called manually.

#
Pool::new

async fn Pool::new(config : PoolConfig) -> Pool raise PgError

#
Pool::start_maintenance

async fn Pool::start_maintenance(self : Pool) -> Unit

Start background maintenance.

Periodically cleans expired idle connections and refills up to min_idle. Runs until Pool::close() is called. Call this once after creating the pool.

When maintenance_interval_sec is 0 (the default) this is a no-op — maintenance happens lazily during acquire() / release() instead.

#
Pool::stats

fn Pool::stats(self : Pool) -> PoolStats

Return a snapshot of pool statistics.

#
Pool::total

fn Pool::total(self : Pool) -> Int

#
PoolConfig

pub(all) struct PoolConfig {
conninfo : String
max_conns : Int
min_idle : Int
max_idle_sec : Int
max_lifetime_sec : Int
health_check : Bool
maintenance_interval_sec : Int
}

Connection pool, inspired by pgxpool.

Concurrency model

MoonBit's async runtime is single-threaded + cooperative. The pool uses @aqueue.Queue(kind=Unbounded) for idle connections (sync try_get / try_put) and get() (async, blocks) when exhausted. A Ref[Int] counter tracks total connections — it is safe because increments happen before the first yield point (connect()).

Basic usage

let pool = Pool::new(PoolConfig::new("postgres://...", max_conns=10)) // Option A: implicit acquire — Pool auto-manages lifecycle let rows = pool.query("SELECT 1") while rows.has_next() { let row = rows.get_row() } rows.close() // PoolRows.close() returns conn to pool let row = pool.query_one("SELECT 42") pool.execute("INSERT ...") |> ignore let tx = pool.begin_tx() tx.commit() // PoolDbTx.commit() returns conn to pool // Option B: explicit acquire — caller manages lifecycle let pc = pool.acquire() defer pc.release() let rows = pc.query("SELECT 1") rows.close() // PoolRows.close() returns conn to pool let row = pc.query_one("SELECT 42") pc.release() // caller releases

#
PoolConfig::new

fn PoolConfig::new(conninfo : String, max_conns? : Int, min_idle? : Int, max_idle_sec? : Int, max_lifetime_sec? : Int, health_check? : Bool, maintenance_interval_sec? : Int) -> PoolConfig

#
PoolConn

pub(all) struct PoolConn {
conn : Connection
pool : Pool
created_at : Int64
released :
Ref
[Bool]
}

impl Closer for PoolConn

#
PoolConn::execute

async fn PoolConn::execute(self : PoolConn, sql : String, params? : Array[&ToValue]) -> ExecResult raise PgError

#
PoolConn::query

async fn PoolConn::query(self : PoolConn, sql : String, params? : Array[&ToValue]) -> &Rows raise PgError

#
PoolConn::query_one

async fn PoolConn::query_one(self : PoolConn, sql : String, params? : Array[&ToValue]) -> Row raise PgError

#
PoolConn::release

fn PoolConn::release(self : PoolConn) -> Unit

#
PoolDbTx

pub(all) struct PoolDbTx {
inner : DbTx
pc : PoolConn
}

Transaction from a pooled connection. Proxies DbTx for queries and releases the connection on commit() / rollback().
impl Tx for PoolDbTx

#
PoolRows

pub(all) struct PoolRows {
inner : &Rows
pc : PoolConn
}

Rows from a pooled connection. Implements Rows — identical API to ConnRows, but close() returns the connection to the pool.
impl Rows for PoolRows

#
PoolStats

pub(all) struct PoolStats {
total_connections : Int
idle_connections : Int
active_connections : Int
acquire_count : Int64
acquire_wait_count : Int64
acquire_wait_duration_ms : Int64
}

Snapshot of pool statistics.

#
Row

pub struct Row {
values : Array[Bytes?]
col_descs : Array[
FieldDescription
]
}

A single row from a query result.

#
Row::columns

Return the column descriptions of this row's result set.

#
Row::get

fn[T : FromValue] Row::get(self : Row, col : Int) -> T raise PgError

Read the value at column col (zero-based index) as T.

#
Row::get_by_name

fn[T : FromValue] Row::get_by_name(self : Row, name : String) -> T raise PgError

Read the value of column name as T.

#
Timestamp

pub(all) struct Timestamp(Int64) derive(Eq,
Debug
)

Microseconds since Unix epoch (1970-01-01 00:00:00 UTC).

#
TxOptions

pub(all) struct TxOptions {
isolation_level : IsolationLevel?
read_only : Bool?
deferrable : Bool?
}

Options for BEGIN — mirrors PostgreSQL's BEGIN ... parameters.

#
TxOptions::default

fn TxOptions::default() -> TxOptions

#
begin_func

async fn[T, B : TxBeginner] begin_func(beginner : B, f : async (&Tx) -> T) -> T raise PgError

Execute a callback inside a transaction.

  1. Call beginner.begin_tx() to start a transaction.
  2. Execute f(tx).
  3. Success → COMMIT, error → ROLLBACK + re-raise.

Works with any TxBeginner: Connection, Pool, PoolConn.

Example

let new_id = begin_func(conn, fn(tx) { let row = tx.query_one("INSERT INTO users (name) VALUES ($1) RETURNING id", params=["alice"]) row.get(0) })

#
build_begin_sql

fn build_begin_sql(opts : TxOptions) -> String

Build a BEGIN SQL string from options.

#
connect

async fn connect(conninfo : String) -> Connection raise PgError

Open a new connection to a PostgreSQL server.

#
get_env

fn get_env(name : String) -> String

Read an environment variable.

#
null

SQL NULL sentinel for use in parameter arrays.

conn.execute("UPDATE t SET email = $1 WHERE id = $2", params=[null, 42])

#
row_from_raw

fn row_from_raw(raw_row : Array[Bytes?], col_descs : Array[
FieldDescription
]) -> Row

Build a Row from raw wire data and column descriptions, respecting the format code in each column. Used by integration tests that request binary result format.