Struct Connection 

pub struct Connection { /* private fields */ }

A connection to a PostgreSQL server.

The connection is established when Connection::connect is called and is closed when the connection is dropped. The connection can be used to execute queries and manage transactions.

Implementations

impl Connection

pub async fn connect(config: &Config) -> Result<Self>

Establishes a new connection to the PostgreSQL server using the given configuration.

This method performs the following steps:

1. Resolves the host and port to a TCP address.
2. Establishes a TCP connection (with optional TLS).
3. Performs the PostgreSQL startup handshake.
4. Authenticates with the server.
5. Collects server parameters until ReadyForQuery.

pub async fn connect_with_retry( config: &Config, retry_policy: &RetryPolicy, ) -> Result<Self>

Connect with automatic retry using the given retry policy.

This is useful for connection establishment that may fail transiently (e.g. the server is temporarily unavailable). The retry policy controls the number of attempts and the delay between them.

pub async fn connect_str(s: &str) -> Result<Self>

Convenience: connect from a connection string (URI or key-value format).

pub fn config(&self) -> &Config

Returns a reference to the configuration used for this connection.

pub async fn set_param(&mut self, key: &str, value: &str) -> Result<()>

Set a runtime parameter on the server.

Sends SET key = value and tracks the change in [SessionState] for automatic re-application on reconnection.

Example

conn.set_param("timezone", "UTC").await?;

pub fn set_reconnect_init_sql(&mut self, sql: impl Into<String>)

Register opaque initialization SQL to replay after a reconnect.

This is intended for idempotent baseline session setup such as pool-level after_connect hooks. The SQL is replayed before tracked session state (for example, set_param() changes) is rebuilt, so later runtime overrides still win.

pub fn clear_reconnect_init_sql(&mut self)

Clear any previously registered reconnect initialization SQL.

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

Return the registered reconnect initialization SQL, if any.

pub fn state(&self) -> ConnectionState

Returns the current connection state.

pub fn is_closed(&self) -> bool

Returns whether the connection is closed or unusable.

pub fn is_idle(&self) -> bool

Returns whether the connection is idle and ready to accept commands.

pub fn server_params(&self) -> &ServerParams

Returns the server parameters collected during connection startup.

pub fn server_version(&self) -> &str

Returns the server version string (e.g. “16.0”).

pub fn process_id(&self) -> i32

Returns the backend process ID.

pub fn secret_key(&self) -> i32

Returns the secret key (used for cancel requests).

pub fn transaction_status(&self) -> TransactionStatus

Returns the current transaction status.

pub fn drain_notifications(&mut self) -> Vec<Notification>

Takes any queued notifications, leaving the queue empty.

pub fn cancel_token(&self) -> CancelToken

Get a cancellation token for this connection.

The token can be sent to another task or thread to cancel a running query. It contains the host, port, process ID, and secret key needed to send an out-of-band cancellation request.

Example

let token = conn.cancel_token();

// In another task:
tokio::spawn(async move {
    tokio::time::sleep(Duration::from_secs(5)).await;
    token.cancel().await.unwrap();
});

pub fn set_notice_handler(&mut self, handler: NoticeHandler)

Set a handler that will be called for every server notice.

The previous handler (if any) is replaced.

pub fn clear_notice_handler(&mut self)

Remove the current notice handler.

pub fn is_tls(&self) -> bool

Returns true if the connection is using TLS.

pub fn tls_info(&self) -> Option<TlsInfo>

Get TLS info if the connection is encrypted.

Returns None if the connection is not using TLS.

pub async fn ping(&mut self) -> Result<()>

Check if the connection is still alive by sending a simple query.

pub fn is_healthy(&self) -> bool

Check connection state without sending a query. Examines the transport and protocol state.

pub async fn reset(&mut self) -> Result<()>

Reset the connection state (clear failed transaction, discard temp objects).

pub fn needs_recovery(&self) -> bool

Whether the connection needs recovery (e.g., a RowStream was dropped before being fully consumed).

When this returns true, the connection may have unread protocol messages in its buffer. Call Connection::recover to drain them and restore the connection to a usable state.

pub async fn recover(&mut self) -> Result<()>

Recover the connection after an incomplete stream consumption.

Reads messages until ReadyForQuery is received, discarding everything. This is needed when a RowStream is dropped before being fully consumed.

pub fn is_alive(&self) -> bool

Check if the connection is believed to be alive.

This is a fast check based on internal state. It does not send a query. For a definitive check, use ping().

pub fn is_stale(&self, threshold: Duration) -> bool

Check if the connection might be broken based on time since last use.

Returns true if the connection hasn’t been confirmed alive in longer than the specified threshold. This is a heuristic — the connection might still be alive, but it’s worth checking before use.

pub fn reconnect_count(&self) -> u32

Get the number of times this connection has been reconnected.

pub fn session_state(&self) -> &SessionState

Get a reference to the session state.

pub async fn reconnect(&mut self) -> Result<()>

Attempt to reconnect this connection.

This closes the current (broken) connection and establishes a new one using the original configuration. If rebuild_session is enabled in the reconnection config, registered reconnect initialization SQL is replayed first and tracked session state (LISTEN channels, GUC parameters) is rebuilt afterwards.

Safety

This should only be called when the connection is known to be broken. Calling this on a live connection will close it and create a new one, which may cause server-side state to be lost.

pub async fn with_retry<T, F, Fut>(&mut self, f: F) -> Result<T>

where F: Fn(&mut Connection) -> Fut, Fut: Future<Output = Result<T>>,

Execute an operation with automatic reconnection and retry.

This is the primary resilience method. It:

1. Executes the operation
2. If the connection is broken, attempts to reconnect and retry
3. If the error is transient (serialization failure, deadlock), retries
4. Respects the configured retry policy (max attempts, backoff)

Example

let result = conn.with_retry(|conn| {
    conn.query_params("SELECT * FROM users WHERE id = $1", &[&user_id])
}).await?;

pub async fn ensure_alive(&mut self) -> Result<()>

Ensure the connection is alive before use.

If the connection is stale (hasn’t been used recently), ping it to verify it’s still alive. If it’s broken, attempt reconnection if configured.

pub async fn close(&mut self) -> Result<()>

Gracefully closes the connection.

Sends a Terminate message (X) to the server and shuts down the underlying transport. After closing, the connection cannot be used for further operations.

pub async fn abort(&mut self)

Force-close the connection without sending a Terminate message.

This is useful when the connection is already known to be broken.

impl Connection

pub async fn copy_in(&mut self, sql: &str) -> Result<CopyIn<'_>>

Start a COPY IN operation.

The sql should be a COPY ... FROM STDIN statement.

Example

let mut copy = conn.copy_in("COPY users (id, name) FROM STDIN").await?;
copy.write_row(&["1", "alice"]).await?;
copy.write_row(&["2", "bob"]).await?;
let rows = copy.finish().await?;

pub async fn copy_out(&mut self, sql: &str) -> Result<CopyOut<'_>>

Start a COPY OUT operation.

The sql should be a COPY ... TO STDOUT statement.

Example

let mut copy = conn.copy_out("COPY users TO STDOUT").await?;
while let Some(chunk) = copy.read_next().await? {
    println!("{}", String::from_utf8_lossy(&chunk));
}

impl Connection

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

Start listening for notifications on a channel.

This executes LISTEN <channel> on the server. After this call, any NOTIFY on the same channel (from any connection) will cause a NotificationResponse message to be sent to this connection.

Example

conn.listen("events").await?;

pub async fn unlisten(&mut self, channel: &str) -> Result<()>

Stop listening on a channel.

This executes UNLISTEN <channel> on the server.

pub async fn unlisten_all(&mut self) -> Result<()>

Stop listening on all channels.

This executes UNLISTEN * on the server.

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

Send a notification on a channel.

This uses pg_notify(channel, payload) which properly handles identifier quoting and payload escaping.

Example

conn.notify("events", "user_logged_in").await?;

pub fn notifications(&mut self) -> Vec<Notification>

Take all buffered notifications from the internal queue.

Notifications can arrive at any time during other operations. They are buffered in an internal queue. This method drains the queue and returns all notifications that have arrived since the last call.

This is a synchronous operation — no I/O is performed.

pub async fn wait_for_notification( &mut self, timeout: Option<Duration>, ) -> Result<Option<Notification>>

Wait for the next notification to arrive.

If a notification is already buffered, it is returned immediately. Otherwise, this method blocks (async) until a notification arrives or the optional timeout expires.

On native tokio builds, timeout semantics are real: the read wait is raced against the requested deadline. On other targets the timeout path remains best-effort and uses an empty query cycle to flush pending notifications.

Example

// Wait up to 5 seconds for a notification
if let Some(notification) = conn.wait_for_notification(Some(Duration::from_secs(5))).await? {
    println!("Got notification on {}: {}", notification.channel, notification.payload);
}

pub async fn wait_for_notification_with_timeout( &mut self, timeout: Duration, ) -> Result<Option<Notification>>

Wait for a notification with a timeout.

If the notification queue already contains a notification, returns it immediately. Otherwise, waits up to timeout for a new notification. Returns Ok(None) if no notification arrives within the timeout.

Example

if let Some(n) = conn.wait_for_notification_with_timeout(Duration::from_secs(5)).await? {
    println!("Got notification on {}: {}", n.channel, n.payload);
}

impl Connection

pub async fn query_cursor( &mut self, sql: &str, params: &[&dyn ToSql], fetch_size: i32, ) -> Result<Cursor<'_>>

Open a cursor for a parameterized query.

The query is parsed and bound to a named portal. The first batch of rows is fetched via Cursor::fetch_next.

Important: Named portals only survive within a transaction block. If no transaction is active, this method automatically begins one so the portal remains valid across fetch_next calls. The transaction is committed when the cursor is closed.

pub async fn query_cursor_stream( &mut self, sql: &str, params: &[&dyn ToSql], fetch_size: i32, ) -> Result<CursorStream<'_>>

Open a streaming cursor for a parameterized query.

Like Connection::query_cursor, but returns a CursorStream that yields rows one at a time instead of in batches. When the current batch is exhausted, the next batch is automatically fetched from the server.

Important: Named portals only survive within a transaction block. If no transaction is active, this method automatically begins one so the portal remains valid. The transaction is committed when the stream is fully consumed or closed.

Example

let mut stream = conn.query_cursor_stream(
    "SELECT id, name FROM users WHERE active = $1",
    &[&true],
    100, // fetch 100 rows at a time
).await?;
while let Some(row) = stream.next().await? {
    let id: i32 = row.get(0)?;
}

impl Connection

pub async fn query_params( &mut self, sql: &str, params: &[&dyn ToSql], ) -> Result<QueryResult>

Execute a parameterized query (one-shot extended query).

Uses the unnamed prepared statement and unnamed portal. Parameters are sent in text format (the server infers types from the SQL). Result columns are requested in binary format where supported.

Wire protocol

The messages Parse, Bind, Execute, and Sync are written into the transport’s write buffer without flushing between them, then a single flush() is issued. This avoids Nagle’s-algorithm-induced deadlocks where small individual writes are buffered by the kernel while the server is waiting for the complete pipeline.

Example

let result = conn
    .query_params("SELECT * FROM users WHERE id = $1", &[&42i32])
    .await?;

pub async fn execute_params( &mut self, sql: &str, params: &[&dyn ToSql], ) -> Result<ExecuteResult>

Execute a parameterized statement that does not return rows.

pub async fn query_params_stream( &mut self, sql: &str, params: &[&dyn ToSql], ) -> Result<RowStream<'_>>

Execute a parameterized query and return a streaming result.

Unlike query_params, this does not buffer all rows. Rows are fetched one at a time as the consumer calls RowStream::next.

Uses the extended query protocol (Parse + Bind + Describe + Execute

• Sync). The ParseComplete, BindComplete, and RowDescription/NoData preamble is consumed eagerly so that parameter-binding errors are surfaced during this call rather than lazily during stream iteration.

Example

let mut stream = conn
    .query_params_stream("SELECT id, name FROM users WHERE age > $1", &[&18i32])
    .await?;
while let Some(row) = stream.next().await? {
    let id: i32 = row.get(0)?;
}

pub async fn query_prepared( &mut self, stmt: &PreparedStatement, params: &[&dyn ToSql], ) -> Result<QueryResult>

Execute a previously prepared statement with parameters.

Parameters are encoded in binary format using the types stored in the PreparedStatement. Results are requested in binary format.

pub async fn execute_prepared( &mut self, stmt: &PreparedStatement, params: &[&dyn ToSql], ) -> Result<ExecuteResult>

Execute a previously prepared statement that does not return rows.

impl Connection

pub fn pipeline(&mut self) -> Pipeline<'_>

Start building a pipeline of extended-query operations.

See Pipeline for details.

impl Connection

pub async fn prepare(&mut self, sql: &str) -> Result<PreparedStatement>

Prepare a statement for repeated execution.

The server parses the SQL, infers parameter types, and returns result column metadata. The returned PreparedStatement can be passed to Connection::query_prepared to execute with parameters.

Example

let stmt = conn.prepare("SELECT * FROM users WHERE id = $1").await?;
let rows = conn.query_prepared(&stmt, &[&42i32]).await?;

pub async fn close_statement(&mut self, stmt: &PreparedStatement) -> Result<()>

Deallocate a prepared statement on the server.

impl Connection

pub async fn query(&mut self, sql: &str) -> Result<QueryResult>

Execute a SQL query that returns rows.

This is a convenience method that collects all rows into a QueryResult. For streaming results one row at a time, use Connection::query_stream instead.

Example

let result = conn.query("SELECT id, name FROM users").await?;
for row in result.iter() {
    let id: i32 = row.get(0)?;
    let name: String = row.get(1)?;
}

pub async fn execute(&mut self, sql: &str) -> Result<ExecuteResult>

Execute a SQL statement that does not return rows.

Returns the number of rows affected where applicable.

pub async fn query_one(&mut self, sql: &str) -> Result<Option<Row>>

Execute a query and return at most one row.

Returns None if the query returns zero rows.

pub async fn query_each<F>(&mut self, sql: &str, f: F) -> Result<CommandTag>

where F: FnMut(Row) -> Result<()>,

Execute a query, invoking f for each row as it arrives.

This avoids buffering all rows in memory, which is useful for large result sets.

pub async fn batch_execute(&mut self, sql: &str) -> Result<Vec<QueryResult>>

Execute multiple statements separated by semicolons.

Returns a QueryResult for each statement that produces one.

impl Connection

pub async fn query_stream(&mut self, sql: &str) -> Result<RowStream<'_>>

Execute a simple query and return a stream of rows.

This is the primary streaming API. Rows are fetched from the server one at a time as the consumer calls next() on the returned stream. Memory usage is O(1) per row regardless of result set size.

Example

let mut stream = conn.query_stream("SELECT id, name FROM users").await?;
while let Some(row) = stream.next().await? {
    let id: i32 = row.get(0)?;
    let name: String = row.get(1)?;
}

pub async fn query_prepared_stream( &mut self, stmt: &PreparedStatement, params: &[&dyn ToSql], ) -> Result<RowStream<'_>>

Execute a prepared statement and return a stream of rows.

pub async fn query_each_async<F, Fut>( &mut self, sql: &str, f: F, ) -> Result<CommandTag>

where F: FnMut(Row) -> Fut, Fut: Future<Output = Result<()>>,

Execute a query and process rows with an async callback (streaming).

Like query_each() but the callback is async, allowing async operations (e.g., writing to another connection) per row.

impl Connection

pub async fn transaction(&mut self) -> Result<Transaction<'_>>

Begin a new transaction.

Example

let mut txn = conn.transaction().await?;
txn.execute("INSERT INTO users (name) VALUES ('alice')").await?;
txn.commit().await?;

pub async fn transaction_with( &mut self, options: &TransactionOptions, ) -> Result<Transaction<'_>>

Begin a new transaction with the given options.

Example

let mut txn = conn.transaction_with(
    TransactionOptions::new()
        .isolation_level(IsolationLevel::Serializable)
        .read_only(true)
).await?;

pub async fn with_transaction<T, F>(&mut self, f: F) -> Result<T>

where F: AsyncFnOnce(&mut Transaction<'_>) -> Result<T>,

Execute an async closure within a transaction.

Commits on Ok, rolls back on Err. This requires Rust 1.85+ (async closures are stable).

Example

let rows: Vec<i32> = conn.with_transaction(async |txn| {
    txn.execute("INSERT INTO nums (v) VALUES (1)").await?;
    let result = txn.query("SELECT v FROM nums").await?;
    let vals: Vec<i32> = result.iter().map(|r| r.get(0).unwrap()).collect();
    Ok(vals)
}).await?;

Trait Implementations

impl Drop for Connection

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.