magmalake postgres

postgres · API

postgres.connection

postgres.connection — the connection, its statements and its transactions.

Connection is the way in to one libpq PGconn. Everything a caller does with a database goes through it: Connection.execute for commands, Connection.query for rows, Connection.prepare for a server-side statement to run more than once, Connection.begin for a transaction, and Connection.copy_in/Connection.copy_out for bulk transfer (the handles those return live in postgres.copy).

Values are never interpolated into SQL. Write $1, $2, ... and pass a Params; libpq sends the values out of band, so there is nothing to escape and no injection surface:

from postgres import Connection, Params

var conn = Connection("postgresql://localhost/app?connect_timeout=5")
_ = conn.execute("CREATE TABLE t (id bigint primary key, name text)")
_ = conn.execute("INSERT INTO t VALUES ($1, $2)",
                 Params().int64(1).text("Alice"))

var res = conn.query("SELECT id, name FROM t WHERE id = $1", Params().int64(1))
for row in res:
    print(row.int64("id"), row.text("name"))

Errors carry their SQLSTATE. Every failure the server reports is raised as a PostgresError formatted by sqlstate.PostgresError.format -- postgres [SQLSTATE 23505] duplicate key value ... -- so a try/except that only has an Error in hand can still branch on sqlstate.sqlstate_of. The structured value is also kept in Connection.last_error, where the is_unique_violation-style predicates are available without any parsing.

Transactions are explicit. Connection.begin issues BEGIN and hands back a Transaction that rolls itself back if it is destroyed without a Transaction.commit -- so an early return or a raised error cannot leave a block half-applied:

var tx = conn.begin()
_ = tx.execute("UPDATE accounts SET balance = balance - $1 WHERE id = $2",
               Params().numeric("10.00").int64(1))
tx.savepoint("maybe")
_ = tx.execute("INSERT INTO audit VALUES ($1)", Params().int64(1))
tx.commit()

A handle keeps the session open. The PGconn is owned jointly by the Connection and by every Statement, Transaction, copy.CopyIn and copy.CopyOut made from it, and it is closed when the last of them is destroyed. That matters because Mojo destroys a value after its last use, not at the end of the block, so this is the ordinary case rather than an exotic one:

var out = conn.copy_out("COPY t TO STDOUT")   # the last mention of `conn`,
for line in out.rows():                       # and the session is still up
    ...

Connection.close is the exception, and it is the point of having it: it ends the session immediately, and every handle still holding the connection then raises SQLSTATE 08006 rather than reaching a freed PGconn.

One connection, one thread. libpq's PGconn is not safe to use from two threads at once, and nothing here adds a lock. Shared ownership is about lifetime, not concurrency.

Notices go to stderr. A NOTICE/WARNING the server raises -- from a RAISE NOTICE, or from CREATE TABLE IF NOT EXISTS finding the table already there -- is printed to standard error by libpq's default notice processor. This tin installs none of its own, so those messages are visible but not capturable; a hook for them is not in v1.

Structs

Statement

struct Statement

A statement prepared on the server, ready to run with new parameters.

Created by Connection.prepare. Preparing pays for parsing and planning once; each Statement.execute or Statement.query then sends only the name and the parameter values.

A prepared statement is session-local: it lives on the connection it was prepared on, until that connection closes or Statement.deallocate runs. Holding a Statement keeps that connection open -- it shares ownership of the PGconn rather than borrowing it -- so a statement may safely outlive the Connection value it came from, which matters because Mojo destroys a value after its last use:

var stmt = conn.prepare("s", "SELECT 1")   # the last mention of `conn`,
var res = stmt.query()                     # and the session is still up

An explicit Connection.close is the exception: it closes the session there and then, and every statement on it raises SQLSTATE 08006 afterwards.

Example:

var stmt = conn.prepare("insert_row", "INSERT INTO t VALUES ($1, $2)")
for i in range(1000):
    _ = stmt.execute(Params().int64(i).text("row"))

Note: Errors raised through a Statement are recorded in Connection.last_error: the statement holds a share of the same cell the connection does.

Implements Movable

__init__

def __init__(out self, cell: ArcPointer[_ConnCell], name: String, sql: String)

Record an already-prepared statement.

Connection.prepare is the only intended caller: it issues PQprepare and checks the status before constructing this.

Args
cell: The shared cell of the connection it was prepared on; a share of it is kept, so the connection stays open at least as long as this statement does.
name: The server-side statement name.
sql: The SQL it was prepared from.

name

def name(self) -> String

The server-side statement name.

sql

def sql(self) -> String

The SQL this statement was prepared from.

execute

def execute(self, params: Params = Params()) -> Int

Run the statement and discard any rows.

Args
params: The values for $1, $2, ....

Raises Error: A sqlstate.PostgresError if the server rejected the command.

query

def query(self, params: Params = Params()) -> Result

Run the statement and return its rows.

Args
params: The values for $1, $2, ....

Raises Error: A sqlstate.PostgresError if the server rejected the command.

deallocate

def deallocate(mut self)

Drop the statement from the server session.

Rarely needed -- closing the connection drops every statement it prepared -- but a long-lived connection that prepares statements dynamically will otherwise accumulate them.

Raises Error: A sqlstate.PostgresError if DEALLOCATE failed, which for an already-deallocated statement is SQLSTATE 26000.

Connection

struct Connection

One connection to a PostgreSQL server.

Opening blocks until the connection succeeds or fails. Set connect_timeout in the conninfo (or on the ConnectionConfig, where it defaults to 10 seconds): without it, a host that silently drops packets blocks for the operating system's TCP timeout, which can be minutes.

The connection is closed when the last thing holding it is destroyed -- the Connection and every Statement, Transaction, copy.CopyIn and copy.CopyOut made from it share ownership of the PGconn -- or at once by Connection.close, which invalidates all of them. Not copyable: a second Connection on one session is what a handle is for.

Example:

var conn = Connection("postgresql://user@localhost/app?connect_timeout=5")
var n = conn.execute("DELETE FROM sessions WHERE expires < now()")
print(n, "sessions expired")

Implements Movable

__init__

def __init__(out self, conninfo: String)

Connect using a libpq conninfo string or URI.

Args
conninfo: Either a URI -- postgresql://user:pass@host:5432/db?sslmode=require -- or a keyword string -- host=localhost dbname=app connect_timeout=5. libpq parses both through the same entry point, and an empty string means "everything from the environment" (PGHOST, PGDATABASE, ...).

Raises Error: A sqlstate.PostgresError with SQLSTATE sqlstate.SQLCLIENT_UNABLE_TO_ESTABLISH (08001) if the connection could not be established, carrying libpq's own explanation.

def __init__(out self, config: ConnectionConfig)

Connect using a typed config.ConnectionConfig.

Args
config: The connection parameters; rendered with config.ConnectionConfig.to_conninfo.

Raises Error: A sqlstate.PostgresError with SQLSTATE 08001 if the connection could not be established.

__deinit__

def __deinit__(deinit self)

Drop this share of the connection.

The PGconn itself is closed by _ConnCell, when the last share of it goes -- which is this one unless a Statement, a Transaction or a COPY handle made from it is still alive, in which case the session stays up for them. Connection.close is how to end it now.

close

def close(mut self)

Close the connection now, whatever else is still holding it. Idempotent.

Every Result already handed out stays valid -- their rows were copied out of libpq's memory. Everything that reaches the server does not: a Statement, Transaction, copy.CopyIn or copy.CopyOut made from this connection raises SQLSTATE 08006 from here on, rather than touching a freed handle.

is_open

def is_open(self) -> Bool

Whether the connection is open and usable.

is_alive

def is_alive(self) -> Bool

Whether the socket is still there, without asking the server.

Between Connection.is_open, which is free but blind, and Connection.ping, which is honest but costs a round trip. This one reads whatever the server has already sent -- a non-blocking PQconsumeInput -- and then asks PQstatus. Nothing is sent, so there is no round trip; but a peer that has gone away has sent something (an ErrorResponse, or just the close), and this is what collects it.

The difference is not academic. A backend killed with pg_terminate_backend, a server restarted underneath a long-lived client, a failover: Connection.is_open reports True through all of them, because libpq's status is a cached opinion that only changes when a command fails. This reports False.

Note: Cheap, not certain. It cannot see a death whose notification has not arrived yet, and it cannot see one that has not happened yet; no local check can. Use it to avoid handing out a connection already known to be dead -- pool.ConnectionPool does exactly that -- and keep handling the error from the next command anyway.

Only meaningful on an idle connection: mid-COPY, or with an unread result outstanding, the bytes waiting on the socket belong to whoever is driving that exchange.

backend_pid

def backend_pid(self) -> Int

The process ID of the backend serving this connection.

Free -- it was read from the server's startup packet, and no round trip is involved. It is the value pg_terminate_backend takes, and it changes across a Connection.reset, which makes it the way to tell a recycled session from a reused one.

reset

def reset(mut self) -> Bool

Reconnect in place: same handle, same parameters, new session.

Blocks for as long as opening would. The PGconn pointer does not change, so every Statement, Transaction and COPY handle sharing this connection stays valid -- but what they are pointed at is a different session: no prepared statements, no temp tables, no open transaction, and a different Connection.backend_pid. A Statement that survives a reset will therefore fail with SQLSTATE 26000, which is the honest answer.

pool.ConnectionPool uses this to recycle a connection whose backend died, rather than paying for a fresh PGconn.

ping

def ping(mut self) -> Bool

Whether the server answers a trivial query right now.

Stronger than Connection.is_open, which only reports libpq's cached view: this actually goes to the server, so it notices a connection that has died since the last command.

server_version

def server_version(self) -> Int

The version of the server on the other end.

Raises Error: Only if libpq itself could not be reached.

transaction_status

def transaction_status(self) -> Int

Whether a transaction block is open on this connection.

Raises Error: Only if libpq itself could not be reached.

last_error

def last_error(self) -> PostgresError

The most recent error raised through this connection.

The structured form of what was raised, so the SQLSTATE predicates are available without parsing the message back:

try:
    _ = conn.execute("INSERT INTO t VALUES (1)")
except:
    if conn.last_error().is_unique_violation():
        ...

execute

def execute(mut self, sql: String, params: Params = Params()) -> Int

Run a command and discard any rows it returned.

With no parameters this goes through PQexec, which accepts several statements separated by semicolons -- handy for a schema script, and the only reason to prefer it. The return value is then the last statement's count, and the first failure aborts the rest. With parameters it goes through PQexecParams, which allows exactly one statement.

Running a SELECT here is allowed: the rows are fetched and dropped, and the count of them is returned. Use Connection.query if you want the rows themselves.

Args
sql: The statement text, with $1-style placeholders if params is non-empty.
params: The parameter values; empty by default.

Raises Error: A sqlstate.PostgresError if the connection is closed or the server rejected the command. The SQLSTATE is recoverable from the message with sqlstate.sqlstate_of, and the whole structured error from Connection.last_error.

query

def query(mut self, sql: String, params: Params = Params()) -> Result

Run a command and return its rows.

Args
sql: The statement text, with $1-style placeholders if params is non-empty. With parameters libpq allows exactly one statement.
params: The parameter values; empty by default.

Raises Error: A sqlstate.PostgresError if the connection is closed or the server rejected the command.

prepare

def prepare(mut self, name: String, sql: String) -> Statement

Prepare sql on the server under name.

Parameter types are left to the server to infer from the statement -- which is what you want in almost every case, and why there is no OID argument. Add a cast ($1::int) where the inference would otherwise land on text.

Args
name: The statement name. "" is libpq's unnamed statement, which the next Connection.prepare on this connection replaces. Any other name persists until the session ends or Statement.deallocate runs, and re-preparing the same name is an error (SQLSTATE 42P05).
sql: The statement text, with $1, $2, ... placeholders.

Raises Error: A sqlstate.PostgresError if the statement did not parse or the name is already taken.

execute_prepared

def execute_prepared(mut self, name: String, params: Params = Params()) -> Int

Run the prepared statement name and discard any rows.

The method Statement.execute is a convenience over; use it directly when the statement was prepared elsewhere -- by another library, or by a PREPARE statement -- and there is no Statement value in hand.

Args
name: The server-side statement name.
params: The values for $1, $2, ....

Raises Error: A sqlstate.PostgresError if the connection is closed, no such statement is prepared (SQLSTATE 26000), or the server rejected the command.

query_prepared

def query_prepared(mut self, name: String, params: Params = Params()) -> Result

Run the prepared statement name and return its rows.

Args
name: The server-side statement name.
params: The values for $1, $2, ....

Raises Error: A sqlstate.PostgresError if the connection is closed, no such statement is prepared, or the server rejected the command.

begin

def begin(mut self) -> Transaction

Issue BEGIN and return the guard that ends the block.

The returned Transaction rolls back when it is destroyed unless Transaction.commit or Transaction.rollback ran first, so the failure mode of forgetting about it is "nothing happened" rather than "half of it happened". Statements inside the block go through the Transaction's own Transaction.execute and Transaction.query, which are the same calls with the same error path.

Raises Error: A sqlstate.PostgresError if the connection is closed, if BEGIN itself failed, or -- SQLSTATE 25001 -- if a transaction block is already open on this connection. PostgreSQL answers a nested BEGIN with a warning and ignores it, which would leave the second guard's COMMIT ending the first one's block; raising is the only way to keep the guard's promise.

copy_in

def copy_in(mut self, sql: String) -> CopyIn

Start a COPY ... FROM STDIN and return the handle to write to.

The fast path for bulk loading: one statement, then rows streamed straight into the table with no per-row parse, plan or round trip. Encode the rows with copyfmt.CopyEncoder and hand them to copy.CopyIn.write_rows, or write already-formatted bytes with copy.CopyIn.write.

Args
sql: The full COPY statement, for example COPY t (a, b) FROM STDIN or COPY t FROM STDIN (FORMAT csv). It takes no parameters -- COPY never has -- so this goes through PQexec.

Raises Error: A sqlstate.PostgresError if the connection is closed, the server rejected the statement (a missing table is 42P01), or -- SQLSTATE 42809 -- the statement was valid but did not start a COPY ... FROM STDIN.

copy_out

def copy_out(mut self, sql: String) -> CopyOut

Start a COPY ... TO STDOUT and return the handle to read from.

Args
sql: The full COPY statement, for example COPY t TO STDOUT or COPY (SELECT ...) TO STDOUT (FORMAT csv).

Raises Error: A sqlstate.PostgresError if the connection is closed, the server rejected the statement, or -- SQLSTATE 42809 -- the statement did not start a COPY ... TO STDOUT.

Transaction

struct Transaction

An open transaction block, and the guard that ends it.

Created by Connection.begin, which issues BEGIN. Run the block's statements through Transaction.execute and Transaction.query -- the same calls as the Connection's, with the same parameters and the same errors -- and finish with Transaction.commit:

var tx = conn.begin()
_ = tx.execute("INSERT INTO orders VALUES ($1)", Params().int64(1))
_ = tx.execute("INSERT INTO items VALUES ($1, $2)",
               Params().int64(1).int32(3))
tx.commit()

Destruction rolls back. If the guard is destroyed without a Transaction.commit -- an early return, a raised error, a forgotten call -- its destructor issues ROLLBACK. The failure mode of losing track of a transaction is therefore "nothing happened", never "half of it happened". Since a with block ends by destroying the guard, that is also what makes this shape work:

with conn.begin() as tx:
    _ = tx.execute("INSERT INTO orders VALUES ($1)", Params().int64(1))
    tx.commit()          # explicit: leaving the block does NOT commit

Leaving that block without the Transaction.commit line -- by falling off the end, by returning, or by raising -- rolls back. There is no implicit commit anywhere in this type.

A failed block cannot be committed. PostgreSQL turns a COMMIT inside a failed transaction into a ROLLBACK and reports success; Transaction.commit raises SQLSTATE 25P02 instead, after issuing the ROLLBACK itself so the connection is left clean. Recover from a failed statement with Transaction.savepoint and Transaction.rollback_to.

Like Statement, this shares ownership of the connection rather than borrowing it: the session stays open for as long as the transaction does, so a Transaction may outlive the Connection value it came from -- and it routinely does, since Mojo destroys a value after its last use. An explicit Connection.close is the exception, and ends the block: every call on the transaction then raises SQLSTATE 08006.

Implements Movable

__init__

def __init__(out self, cell: ArcPointer[_ConnCell])

Issue BEGIN on the cell's connection.

Connection.begin is the only intended caller: it checks that the connection is open and that no block is open already.

Args
cell: The shared cell of the connection to open a transaction on; a share of it is kept, so the connection stays open at least as long as this transaction does.

Raises Error: A sqlstate.PostgresError if BEGIN failed.

__deinit__

def __deinit__(deinit self)

Roll back unless the block was already committed or rolled back.

Unconditional: a ROLLBACK is accepted in every transaction state, including the failed one where nothing else is, and it is the only statement that leaves the connection idle again. Errors are swallowed -- a destructor has nobody to raise to -- and a connection that has been closed has already lost the transaction, so nothing is sent.

__enter__

def __enter__(deinit self) -> Self

Hand the guard to with conn.begin() as tx:.

The guard is transferred into the block rather than borrowed, so it is destroyed -- and therefore rolled back, unless Transaction.commit ran -- when the block ends, however it ends.

is_open

def is_open(self) -> Bool

Whether the block is still open.

execute

def execute(mut self, sql: String, params: Params = Params()) -> Int

Run a command inside the block and discard any rows.

Connection.execute's behaviour exactly, including the PQexec path for a parameterless call and the multi-statement strings that allows.

Args
sql: The statement text, with $1-style placeholders if params is non-empty.
params: The parameter values; empty by default.

Raises Error: A sqlstate.PostgresError if the transaction is finished or the server rejected the command. A rejected statement puts the block in the failed state, where nothing but a rollback -- whole, or to a savepoint -- will be accepted.

query

def query(mut self, sql: String, params: Params = Params()) -> Result

Run a command inside the block and return its rows.

Args
sql: The statement text, with $1-style placeholders if params is non-empty.
params: The parameter values; empty by default.

Raises Error: A sqlstate.PostgresError if the transaction is finished or the server rejected the command.

savepoint

def savepoint(mut self, name: String)

Mark a point inside the block to come back to.

The way to survive a failed statement without losing the whole transaction: set a savepoint, try the statement, and Transaction.rollback_to the savepoint if it fails. The block is usable again afterwards.

Args
name: The savepoint name, spliced into the SQL as a quoted identifier -- SAVEPOINT takes an identifier, and identifiers cannot be parameters. Any embedded double quotes are doubled, so a name can never end the quoting early. Re-using a name is allowed: the newer savepoint hides the older one, which is still there after a Transaction.release.

Raises Error: A sqlstate.PostgresError if the transaction is finished, or if the block has already failed -- a savepoint cannot be set in the failed state, only rolled back to.

rollback_to

def rollback_to(mut self, name: String)

Undo everything after the savepoint name, keeping the block open.

Also clears the failed state, which is the point: after this the transaction accepts statements again and can still be committed.

Args
name: The savepoint name, quoted as an identifier.

Raises Error: A sqlstate.PostgresError if the transaction is finished, or if there is no such savepoint (SQLSTATE 3B001).

release

def release(mut self, name: String)

Forget the savepoint name, keeping everything done since it.

The opposite of Transaction.rollback_to: the work stays, and the savepoint -- along with every savepoint set after it -- is gone.

Args
name: The savepoint name, quoted as an identifier.

Raises Error: A sqlstate.PostgresError if the transaction is finished, or if there is no such savepoint (SQLSTATE 3B001).

commit

def commit(mut self)

Commit the block, making everything in it permanent.

The destructor becomes a no-op afterwards, and the Transaction is finished: any further use raises.

Raises Error: A sqlstate.PostgresError if the transaction is already finished, if COMMIT itself failed, or -- SQLSTATE 25P02, transaction is aborted -- if a statement inside the block failed and was not rolled back to a savepoint. PostgreSQL would answer that COMMIT with ROLLBACK and report success, so a caller who did not check every statement would believe the block had been committed; the ROLLBACK is issued here too, leaving the connection idle, but the outcome is reported rather than hidden.

rollback

def rollback(mut self)

Roll the block back, discarding everything in it.

Accepted in every state, including the failed one. The destructor becomes a no-op afterwards, and the Transaction is finished: any further use raises.

Raises Error: A sqlstate.PostgresError if the transaction is already finished, or if ROLLBACK itself failed -- which in practice means the connection has gone.

Functions

drain_results

def drain_results(conn: Int, cell: ArcPointer[_ConnCell], sql: String) -> Int

Collect every outstanding result, then report on the first one.

This is how a COPY ends. LibpqFFI.PQgetResult must be called until it returns 0 or the connection stays busy and every later command fails, so the drain happens first, unconditionally -- only then is the first result's status examined and, if it failed, raised. A malformed COPY row therefore surfaces as the server's own error (22P02 with its DETAIL/CONTEXT) on a connection that is already clean.

Args
conn: The connection handle.
cell: The connection's shared cell.
sql: The statement text, attached to any error raised.

Raises Error: A sqlstate.PostgresError if the first result reports a failure.

drain_results_quietly

def drain_results_quietly(conn: Int)

Collect and discard every outstanding result, reporting nothing.

The destructor's half of drain_results: a handle that is dropped rather than finished still has to leave the connection idle, and a destructor has nobody to raise to.

Args
conn: The connection handle.

discard_copy_quietly

def discard_copy_quietly(conn: Int, status: Int)

End whatever COPY status says is in progress, and drain it, silently.

A COPY holds the connection until it is finished one way or another, and the way out depends on the direction: a COPY FROM STDIN is ended by telling libpq the stream failed, while a COPY TO STDOUT can only be consumed -- PostgreSQL offers no cancel for it. This does whichever applies and then drains the results, so the connection comes back idle.

Used by the COPY destructors, which have nobody to raise to, and by Connection.copy_in/Connection.copy_out when the statement started a COPY in the other direction.

Args
conn: The connection handle.
status: _ffi.PGRES_COPY_IN, _ffi.PGRES_COPY_OUT or _ffi.PGRES_COPY_BOTH -- the state the connection is in. Any other value only drains.

Aliases

comptime IN_FAILED_SQL_TRANSACTION = StringSpan("25P02")

The block has already failed, so COMMIT would silently roll back.

comptime ACTIVE_SQL_TRANSACTION = StringSpan("25001")

A transaction block is already open on this connection.

comptime NO_ACTIVE_SQL_TRANSACTION = StringSpan("25P01")

The transaction is over -- committed or rolled back already.

comptime WRONG_OBJECT_TYPE = StringSpan("42809")

The statement was not the COPY the call needed it to be.