Struct PgConnection 

pub struct PgConnection { /* private fields */ }

An asynchronous PostgreSQL connection backed by tokio-postgres (Pure Rust, no libpq).

The inner Client is wrapped in Arc<Mutex<_>> to allow the Connection trait’s &self transaction method to obtain exclusive mutable access.

Clone is cheap: both Arcs inside are reference-counted.

Statement cache

A second Arc<std::sync::Mutex<HashMap<…>>> holds compiled tokio_postgres::Statement handles keyed on SQL text. The std::sync::Mutex (not tokio::sync::Mutex) is intentional: we only hold the lock during purely synchronous HashMap lookups, never across an await.

Reconnection

When a PgConnection is created via PgConnection::connect, the original connection string and TLS mode are stored internally. Call PgConnection::reconnect to obtain a brand-new PgConnection built from the same parameters — useful after a broken-pipe error. The caller is responsible for replacing their handle with the returned value.

Implementations

impl PgConnection

pub fn from_client(client: Client) -> Self

Construct a PgConnection from an existing tokio_postgres::Client.

Useful when the Client was obtained outside of OxiSQL (e.g., from a custom connection pool or test harness). A fresh, empty statement cache is created automatically.

Example

let (client, conn) = tokio_postgres::connect(
    "host=localhost user=postgres",
    tokio_postgres::NoTls,
).await?;
tokio::spawn(async move { let _: Result<_, _> = conn.await; });

let pg_conn = oxisql_postgres::PgConnection::from_client(client);

pub async fn connect(conn_str: &str, tls: TlsMode) -> Result<Self, PgError>

Connect to a PostgreSQL server using the given connection string.

conn_str follows the tokio-postgres format, e.g.: "host=localhost port=5432 user=postgres password=secret dbname=mydb".

TLS

Pass TlsMode::Disabled for a plain-text connection (suitable for localhost or trusted network). Pass TlsMode::Rustls with a ClientConfig built from OxiTLS for encrypted connections.

pub async fn connect_with_timeout( conn_str: &str, tls: TlsMode, timeout: Duration, ) -> Result<Self, PgError>

Connect to a PostgreSQL server with a maximum wait duration.

Wraps PgConnection::connect with tokio::time::timeout. If the connection is not established within timeout, returns PgError::Timeout rather than hanging indefinitely.

Errors

Returns PgError::Timeout if timeout elapses before the server acknowledges the connection. All other error conditions are propagated unchanged from PgConnection::connect.

Example

use std::time::Duration;
use oxisql_postgres::{PgConnection, TlsMode};

let conn = PgConnection::connect_with_timeout(
    "host=localhost user=postgres",
    TlsMode::Disabled,
    Duration::from_secs(5),
).await?;

pub async fn connect_skip_verify(conn_str: &str) -> Result<Self, PgError>

Connect with TLS that skips server certificate verification.

Security Warning

This method accepts any server certificate without validation. It is vulnerable to man-in-the-middle attacks and should only be used in development or testing environments.

Errors

Returns PgError::Tls if the TLS config cannot be built, or a PgError::Postgres / PgError::Timeout on connection failure.

Example

let conn = oxisql_postgres::PgConnection::connect_skip_verify(
    "host=localhost user=postgres",
).await?;

pub async fn connect_with_ca( conn_str: &str, ca_pem: Vec<u8>, ) -> Result<Self, PgError>

Connect with TLS using a custom CA certificate in PEM format.

The provided ca_pem bytes are parsed as a PEM bundle and added to the trust store (in addition to the Mozilla WebPKI root certificates).

Errors

Returns PgError::Tls if the PEM cannot be parsed or is empty, or a PgError::Postgres / PgError::Timeout on connection failure.

Example

let ca_pem = std::fs::read("/path/to/ca.crt")?;
let conn = oxisql_postgres::PgConnection::connect_with_ca(
    "host=db.example.com user=postgres",
    ca_pem,
).await?;

pub fn server_version(&self) -> Option<&str>

Return the PostgreSQL server version string reported during the connection handshake, if known.

This is the raw server_version runtime parameter as reported by the server in a ParameterStatus message during startup (e.g. "16.4 (Debian 16.4-1.pgdg120+1)") — not a parsed/normalized version number. It is captured once, in PgConnection::connect, before the tokio_postgres::Connection driver is spawned away; tokio-postgres does not expose this parameter on Client itself.

Returns None when this connection was created via PgConnection::from_client (no Connection driver was available to read the parameter from) or in the practically-impossible case that the server did not report server_version at all.

Example

use oxisql_postgres::{PgConnection, TlsMode};

let conn = PgConnection::connect("host=localhost user=postgres", TlsMode::Disabled).await?;
if let Some(version) = conn.server_version() {
    println!("connected to PostgreSQL {version}");
}

impl PgConnection

pub fn is_read_only_query(sql: &str) -> bool

Return true if the SQL is a read-only query (SELECT, EXPLAIN, SHOW, …).

Uses oxisql_parse::parse_one to parse the statement and oxisql_parse::is_read_only for AST-level classification. This can be used to route read queries to a replica and write queries to a primary.

Returns false when the SQL cannot be parsed or when the statement modifies data or schema.

pub fn normalize_query(sql: &str) -> String

Return the normalized canonical form of a SQL query for use as a cache key.

Uses oxisql_parse::normalize to collapse whitespace, strip comments, and upper-case keywords. The result is stable across cosmetically different but semantically equivalent SQL strings, making it suitable as a prepared-statement cache key.

impl PgConnection

pub async fn copy_in_text( &self, table: &str, columns: &[&str], rows: impl Iterator<Item = Vec<String>>, ) -> Result<u64, PgError>

Bulk-insert rows into table using the PostgreSQL COPY … FROM STDIN protocol (text/TSV format).

Each element of rows must contain exactly columns.len() string values. Values are automatically escaped for the PostgreSQL text COPY format. Pass "\\N" (backslash-N) to represent a SQL NULL.

Returns the number of rows inserted as reported by PostgreSQL.

Errors

Returns PgError::Copy if columns is empty, if a row has the wrong number of fields, or if the underlying COPY protocol fails.

pub async fn copy_out_text( &self, table: &str, columns: &[&str], ) -> Result<Vec<Vec<String>>, PgError>

Extract rows from table using the PostgreSQL COPY … TO STDOUT protocol (text/TSV format).

Returns a Vec<Vec<String>> where each inner vec contains the field values for one row, with escape sequences decoded. The literal "\\N" in the returned data represents a SQL NULL.

Errors

Returns PgError::Copy if columns is empty or if the underlying COPY protocol fails.

pub async fn notify(&self, channel: &str, payload: &str) -> Result<(), PgError>

Send a NOTIFY command to channel with an optional payload.

The channel name must consist only of ASCII alphanumeric characters and underscores. The payload is automatically single-quote–escaped before inclusion in the SQL string.

Errors

Returns PgError::Notify if the channel name is invalid or if the underlying query fails.

pub async fn listen(&self, channel: &str) -> Result<NotificationStream, PgError>

Register LISTEN on channel and return a notify::NotificationStream.

Requires that this connection was created via PgConnection::connect. Connections obtained via PgConnection::from_client return PgError::Notify because no tokio_postgres::Connection driver is available to intercept incoming notifications.

Errors

Returns PgError::Notify if the channel name is invalid, if this connection was created via from_client, or if the LISTEN command fails.

pub fn pipeline(&self) -> PgPipeline

Create a new PgPipeline for batched query dispatch.

Queue queries with PgPipeline::add_execute and PgPipeline::add_query, then send all at once with PgPipeline::finish.

pub async fn cancel_token(&self) -> PostgresCancelToken

Return a PostgresCancelToken that can be used to interrupt the currently running query on this connection.

The token is Send + Sync and may be moved to another task. Call PostgresCancelToken::cancel_query with the appropriate TLS connector to send a cancel request to the server.

Note

Because PgConnection wraps the client in an Arc<Mutex<_>>, acquiring the cancel token requires a momentary lock on the inner mutex. The lock is held only for the duration of the synchronous cancel_token() call on the underlying tokio_postgres::Client, never across any await.

Example

use oxisql_postgres::{PgConnection, TlsMode};

let conn = PgConnection::connect("host=localhost user=postgres", TlsMode::Disabled).await?;
let token = conn.cancel_token().await;
// Spawn a task that cancels after a short delay:
tokio::spawn(async move {
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    let _ = token.cancel_query(tokio_postgres::NoTls).await;
});

pub async fn reconnect(&self) -> Result<PgConnection, PgError>

Establish a fresh connection using the same URI and TLS mode as this connection.

Returns a brand-new PgConnection with an empty statement cache. The caller should replace their handle with the returned value after a broken-pipe or connection-lost error.

Errors

Returns PgError::Notify (repurposed as a “no URI” sentinel) when this connection was created via PgConnection::from_client and no original connection string was recorded.

Returns PgError::Postgres on connection failure.

pub async fn query_binary( &self, sql: &str, params: &[&dyn ToSqlValue], ) -> Result<Vec<Row>, PgError>

Execute a query using the binary protocol for better throughput.

Uses tokio-postgres’s query_typed, which sends explicit Postgres type annotations alongside each parameter value. The server then returns result columns in binary format (format code 1), bypassing the text-serialization step used by Connection::query. This reduces per-row encoding/decoding overhead on both client and server, improving throughput for large result sets.

Parameter types are inferred from the ToSqlValue variants:

OxiSQL variant	Declared Postgres type
Bool	BOOL
I64	INT8
F64	FLOAT8
Text	TEXT
Blob	BYTEA
Null / extended types	UNKNOWN (server infers from context)

Note: parameters are declared as INT8/FLOAT8/TEXT rather than narrower types (INT4, FLOAT4, etc.). If a column has a narrower type and no implicit cast is available, prefer Connection::query which lets the server apply type inference after prepare.

Errors

Returns PgError::Postgres on connection or server-side errors. Returns PgError::TypeConversion if a result column cannot be decoded.

Example

use oxisql_postgres::{PgConnection, TlsMode};

let conn = PgConnection::connect("host=localhost user=postgres", TlsMode::Disabled).await?;
let rows = conn.query_binary("SELECT id, name FROM users WHERE id = $1", &[&42_i64]).await?;

pub async fn describe( &self, sql: &str, ) -> Result<Vec<ColumnDescription>, PgError>

Return column metadata for the given SQL without executing it.

Uses client.prepare(sql) to obtain the server-reported column descriptions. Each ColumnDescription carries the column name and PostgreSQL type name (e.g. "int4", "text", "uuid").

nullable is always true; the PostgreSQL extended-query protocol does not carry nullability information in statement descriptions.

Errors

Returns PgError::Postgres if the server rejects the SQL (syntax error, unknown table, etc.).

Trait Implementations

impl Clone for PgConnection

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Connection for PgConnection

fn query_stream<'a>( &'a self, sql: &'a str, params: &'a [&'a dyn ToSqlValue], ) -> Pin<Box<dyn Stream<Item = Result<Row, OxiSqlError>> + Send + 'a>>

Stream query results row-by-row from PostgreSQL.

Acquires a lock on the client, runs client.query() (which fetches all rows in a single round-trip), then yields each converted row individually. This is equivalent to the default trait implementation but avoids going through async_trait’s boxing.

fn prepare<'life0, 'life1, 'async_trait>( &'life0 self, sql: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn PreparedStatement + '_>, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Compile a SQL statement for repeated execution.

The compiled Statement is cached by SQL text: subsequent calls with the same SQL return a clone of the cached handle without any server round-trip.

fn execute<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, sql: &'life1 str, params: &'life2 [&'life3 dyn ToSqlValue], ) -> Pin<Box<dyn Future<Output = Result<u64, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Execute a DML/DDL statement and return the number of rows affected.

fn query<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, sql: &'life1 str, params: &'life2 [&'life3 dyn ToSqlValue], ) -> Pin<Box<dyn Future<Output = Result<Vec<Row>, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Execute a SELECT statement and return the result rows.

fn transaction<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn Transaction + '_>, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Begin a transaction, returning a handle that can be committed or rolled back.

fn execute_batch<'life0, 'life1, 'async_trait>( &'life0 self, sql: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<u64, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Execute multiple semicolon-separated SQL statements in a single call.

fn ping<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lightweight connectivity check.

fn tables<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<TableInfo>, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List all tables visible to the current connection.

fn columns<'life0, 'life1, 'async_trait>( &'life0 self, table: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<ColumnInfo>, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

List all columns of the named table.

fn indexes<'life0, 'life1, 'async_trait>( &'life0 self, table: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<IndexInfo>, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

List all indexes defined on the named table.

fn foreign_keys<'life0, 'life1, 'async_trait>( &'life0 self, table: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<ForeignKeyInfo>, OxiSqlError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

List all foreign-key constraints on the named table.

fn execute_named<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, sql: &'life1 str, params: &'life2 [(&'life3 str, &'life4 dyn ToSqlValue)], ) -> Pin<Box<dyn Future<Output = Result<u64, OxiSqlError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, Self: 'async_trait,

Execute a SQL statement with named parameters.

fn query_named<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, sql: &'life1 str, params: &'life2 [(&'life3 str, &'life4 dyn ToSqlValue)], ) -> Pin<Box<dyn Future<Output = Result<Vec<Row>, OxiSqlError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, Self: 'async_trait,

Execute a SQL query with named parameters and return the result rows.

fn last_warnings(&self) -> Vec<SqlWarning>

Return any SQL warnings generated by the most recently executed statement.