Sign in

    moonpg

    A pure MoonBit PostgreSQL client, wire protocol from scratch.

    postgres
    postgresql
    wire-protocol
    database
    Download zip
    Author
    Version
    0.10.0
    License
    Apache-2.0
    Last updated
    11 days ago
    Downloads
    154

    #jaredzhou/moonpg

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

    #Features

    • Query & Fetch — query/query_one/execute for manual control, fetch/fetch_one with FromRow for typed auto-mapping to structs or tuples.
    • Type-safe codec — ToValue 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.
    • Transactions — begin_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.
    • TLS — sslmode 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] = conn.fetch("SELECT name FROM users ORDER BY id")
    let count : Int = conn.fetch_one("SELECT COUNT(*) FROM users")

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

    // query — manual iteration (defer rows.close())
    let rows = conn.query("SELECT id, name FROM users")
    defer rows.close()
    while rows.has_next() {
    let row = rows.get_row()
    let id : Int = row.get(0)
    let name : String = row.get_by_name("name")
    }

    #Query

    #fetch / fetch_one

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

    They're declared in the Fetching trait — a non-object-safe companion to QueryExecutor, since fetch is generic in T — with a single shared implementation, and are available directly on Connection, Pool, PoolConn, DbTx, PoolDbTx, and &Tx (the type returned by begin_tx). The explicit &QueryExecutor::fetch(x, ...) form keeps working too.

    let conn = @moonpg.connect(conninfo)

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

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

    // Tuples — no struct needed
    let pairs : Array[(Int, String)] = conn.fetch("SELECT id, name FROM users")
    let (id, name, email) : (Int, String, String) = conn.fetch_one(
    "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] = conn.fetch("SELECT id, name, email FROM users")

    #query / query_one

    query returns a &Rows iterator. Always defer rows.close() to drain results and (for pool connections) return the connection — this also runs close() on decode errors.

    // query — iterate rows manually
    let rows = conn.query("SELECT id, name FROM users")
    defer rows.close()
    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}")
    }

    // 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] = pool.fetch("SELECT name FROM users")
    pool.execute("INSERT INTO users (name) VALUES ($1)", params=["bob"]) |> ignore

    // Manual query with defer
    let rows = pool.query("SELECT id FROM users")
    defer rows.close()
    while rows.has_next() { ... }

    // 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
    tx.fetch_one("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 = tx.fetch_one("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 = conn.fetch_one("SELECT COUNT(*) FROM users")
    let name : String? = conn.fetch_one("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] = conn.fetch("SELECT id, name, email FROM users")

    // Tuple impls (2–10)
    let pairs : Array[(Int, String)] = conn.fetch("SELECT id, name FROM users")
    let (id, name, email) : (Int, String, String) = conn.fetch_one(
    "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.

    Fetching

    pub(open) trait Fetching : QueryExecutor {
    async fn[T : FromRow] fetch(Self, String, params? : Array[&ToValue]) -> Array[T] raise PgError = _
    async fn[T : FromRow] fetch_one(Self, String, params? : Array[&ToValue]) -> T raise PgError = _
    }

    Types that can run a SELECT and map its rows into T.

    fetch/fetch_one are polymorphic in T, so they cannot be declared in QueryExecutor: a polymorphic method makes a trait unusable as an object, and &QueryExecutor — and therefore &Tx — is part of this API. They live in this separate, deliberately non-object-safe trait instead, and get a single shared implementation (see impl Fetching with ... below).

    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.

    let rows = conn.query("SELECT id FROM users")
    defer rows.close()
    while rows.has_next() {
    let row = rows.get_row()
    let v : Int = row.get(0)
    }

    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
    )

    PgError::to_repr

    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

    ConnRows::close

    async fn ConnRows::close(self : ConnRows) -> Unit raise PgError

    ConnRows::get_row

    fn ConnRows::get_row(self : ConnRows) -> Row raise PgError

    ConnRows::has_next

    async fn ConnRows::has_next(self : ConnRows) -> Bool raise PgError

    ConnStatus

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

    Connection status.

    ConnStatus::equal

    fn ConnStatus::equal(ConnStatus, ConnStatus) -> Bool

    ConnStatus::not_equal

    fn ConnStatus::not_equal(x : ConnStatus, y : ConnStatus) -> Bool

    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::begin_tx

    async fn Connection::begin_tx(self : Connection, opts? : TxOptions) -> &Tx raise PgError

    Connection::close

    fn Connection::close(self : Connection) -> Unit

    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::execute

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

    Connection::fetch

    async fn[T : FromRow] Connection::fetch(self : Connection, sql : String, params? : Array[&ToValue]) -> Array[T] raise PgError

    Connection::fetch_one

    async fn[T : FromRow] Connection::fetch_one(self : Connection, sql : String, params? : Array[&ToValue]) -> T raise PgError

    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::query

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

    Connection::query_one

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

    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 Fetching for DbTx
    impl Tx for DbTx

    DbTx::commit

    async fn DbTx::commit(self : DbTx) -> Unit raise PgError

    DbTx::execute

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

    DbTx::fetch

    async fn[T : FromRow] DbTx::fetch(self : DbTx, sql : String, params? : Array[&ToValue]) -> Array[T] raise PgError

    DbTx::fetch_one

    async fn[T : FromRow] DbTx::fetch_one(self : DbTx, sql : String, params? : Array[&ToValue]) -> T raise PgError

    DbTx::query

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

    DbTx::query_one

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

    DbTx::rollback

    async fn DbTx::rollback(self : DbTx) -> Unit raise PgError

    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 Fetching for Pool
    impl TxBeginner for Pool

    Pool::acquire

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

    Pool::begin_tx

    async fn Pool::begin_tx(self : Pool, opts? : TxOptions) -> &Tx raise PgError

    Pool::close

    fn Pool::close(self : Pool) -> Unit

    Pool::execute

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

    Pool::fetch

    async fn[T : FromRow] Pool::fetch(self : Pool, sql : String, params? : Array[&ToValue]) -> Array[T] raise PgError

    Pool::fetch_one

    async fn[T : FromRow] Pool::fetch_one(self : Pool, sql : String, params? : Array[&ToValue]) -> T 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::query

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

    Pool::query_one

    async fn Pool::query_one(self : Pool, sql : String, params? : Array[&ToValue]) -> Row 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") defer rows.close() // PoolRows.close() returns conn to pool while rows.has_next() { let row = rows.get_row() } 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") defer 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::begin_tx

    async fn PoolConn::begin_tx(self : PoolConn, opts? : TxOptions) -> &Tx raise PgError

    PoolConn::close

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

    PoolConn::execute

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

    PoolConn::fetch

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

    PoolConn::fetch_one

    async fn[T : FromRow] PoolConn::fetch_one(self : PoolConn, sql : String, params? : Array[&ToValue]) -> T 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

    PoolDbTx::commit

    async fn PoolDbTx::commit(self : PoolDbTx) -> Unit raise PgError

    PoolDbTx::execute

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

    PoolDbTx::fetch

    async fn[T : FromRow] PoolDbTx::fetch(self : PoolDbTx, sql : String, params? : Array[&ToValue]) -> Array[T] raise PgError

    PoolDbTx::fetch_one

    async fn[T : FromRow] PoolDbTx::fetch_one(self : PoolDbTx, sql : String, params? : Array[&ToValue]) -> T raise PgError

    PoolDbTx::query

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

    PoolDbTx::query_one

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

    PoolDbTx::rollback

    async fn PoolDbTx::rollback(self : PoolDbTx) -> Unit raise PgError

    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

    PoolRows::close

    async fn PoolRows::close(self : PoolRows) -> Unit raise PgError

    PoolRows::get_row

    fn PoolRows::get_row(self : PoolRows) -> Row raise PgError

    PoolRows::has_next

    async fn PoolRows::has_next(self : PoolRows) -> Bool raise PgError

    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).

    Timestamp::equal

    fn Timestamp::equal(Timestamp, Timestamp) -> Bool

    Timestamp::from_row

    fn Timestamp::from_row(r : Row) -> Timestamp raise PgError

    Timestamp::from_value

    Timestamp::not_equal

    fn Timestamp::not_equal(x : Timestamp, y : Timestamp) -> Bool

    Timestamp::to_value

    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.