Struct DoConnection 

pub struct DoConnection { /* private fields */ }

A connection to a Durable Object’s SQL storage.

There is no socket behind it and nothing to pool: the database lives in the Durable Object itself. Build one from the object’s storage with DoConnection::new, in its constructor or per request.

Both &mut DoConnection and &DoConnection are sqlx executors, so a connection kept in the object’s struct serves every request through &self.

Implementations

impl DoConnection

pub fn execute_batch<'q, I>( &self, queries: I, ) -> impl Future<Output = Result<Vec<DoQueryResult>, Error>> + Send + '_

where I: IntoIterator<Item = Query<'q, Do, DoArguments>>,

Runs queries in order and atomically: if any statement fails, none of them take effect.

This is the Durable Object answer to a transaction, which sql.exec does not allow (see DoTransactionManager). The statements run inside transaction, and a failing one rolls the others back. Returns what each statement did, in the order given; for the rows each statement returns as well, see fetch_batch.

let results = conn
    .execute_batch([
        sqlx::query("UPDATE accounts SET balance = balance - ?1 WHERE id = ?2")
            .bind(amount)
            .bind(from),
        sqlx::query("UPDATE accounts SET balance = balance + ?1 WHERE id = ?2")
            .bind(amount)
            .bind(to),
    ])
    .await?;

Errors

When binding an argument fails, in which case nothing ran, or when any statement fails – in which case the whole batch was rolled back.

pub fn fetch_batch<'q, I>( &self, queries: I, ) -> impl Stream<Item = Result<DoBatchResult, Error>> + Send + '_

where I: IntoIterator<Item = Query<'q, Do, DoArguments>>,

Runs queries in order and atomically, as execute_batch does, and streams each statement’s rows along with what it did, in the order given.

The shape sqlc’s :batchexec, :batchone and :batchmany queries take, each a stream with one item per statement – but atomic. So unlike a loop of queries it is not lazy: the whole batch runs when the stream is first polled, and a failure is the one item it yields, with nothing applied.

use futures_util::TryStreamExt;
use sqlx::Row;

let names: Vec<String> = conn
    .fetch_batch([1_i64, 2, 3].map(|id| {
        sqlx::query("SELECT name FROM users WHERE id = ?").bind(id)
    }))
    .map_ok(|result| result.rows()[0].get("name"))
    .try_collect()
    .await?;

Errors

The stream’s one item is an error when binding an argument fails, in which case nothing ran, or when any statement fails – in which case the whole batch was rolled back.

impl DoConnection

pub fn new(storage: Storage) -> Self

Wraps a Durable Object’s storage: state.storage().

pub fn storage(&self) -> &Storage

The storage this connection wraps, for anything this crate does not cover – the key-value API, alarms – while keeping the connection.

pub fn sql(&self) -> &SqlStorage

The SQL API this connection runs queries through.

pub fn into_storage(self) -> Storage

The storage this connection wraps, giving up the connection. See storage to borrow it instead.

impl DoConnection

pub fn transaction<F, Fut, T, E>( &self, callback: F, ) -> impl Future<Output = Result<T, E>> + Send + '_

where F: FnOnce(DoTransaction) -> Fut + 'static, Fut: Future<Output = Result<T, E>> + 'static, T: 'static, E: From<Error> + 'static,

Runs callback in a transaction: it commits if the callback returns Ok, and rolls back every write the callback made if it returns Err.

A real interactive transaction, unlike a batch: the callback reads its own writes, and one statement can use what the one before it returned. It runs inside Storage::transaction, and the callback’s value – or its error, as it returned it – is what this resolves to.

let id = conn
    .transaction(move |tx| async move {
        let id: i64 = sqlx::query_scalar("INSERT INTO users (email) VALUES (?) RETURNING id")
            .bind(email)
            .fetch_one(&tx)
            .await?;
        sqlx::query("INSERT INTO posts (user_id, title) VALUES (?, 'hello')")
            .bind(id)
            .execute(&tx)
            .await?;
        Ok::<_, sqlx::Error>(id)
    })
    .await?;

The callback owns what it uses

Storage::transaction takes a 'static callback, so it cannot borrow the connection or anything else from the caller: move what it needs in, and run its queries through the DoTransaction it is handed. That is also why this is a callback and not sqlx’s begin(), which borrows the connection – begin() still fails.

It holds the whole object

The Durable Object delivers no other request until the transaction ends – even while the callback awaits something that is not storage. That is what keeps other requests’ writes out of it, and why it should be short: an awaited fetch() inside it stalls every caller of the object, and a request to the object itself would never be answered.

Errors

The callback’s own error, after rolling back; or, converted into E, an error from storage when the transaction could not commit.

Trait Implementations

impl<'c> Acquire<'c> for &'c mut DoConnection

type Database = Do

type Connection = &'c mut <Do as Database>::Connection

fn acquire(self) -> BoxFuture<'c, Result<Self::Connection, Error>>

fn begin(self) -> BoxFuture<'c, Result<Transaction<'c, Do>, Error>>

impl Connection for DoConnection

fn begin( &mut self, ) -> impl Future<Output = Result<Transaction<'_, Do>, Error>> + Send + '_

Always fails – see DoTransactionManager.

type Database = Do

type Options = DoConnectOptions

fn close(self) -> impl Future<Output = Result<(), Error>> + Send + 'static

Explicitly close this database connection.

fn ping(&mut self) -> impl Future<Output = Result<(), Error>> + Send + '_

Checks if a connection to the database is still valid.

fn shrink_buffers(&mut self)

Restore any buffers in the connection to their default capacity, if possible.

fn begin_with( &mut self, statement: impl SqlSafeStr, ) -> impl Future<Output = Result<Transaction<'_, Self::Database>, Error>> + Send

where Self: Sized,

Begin a new transaction with a custom statement.

fn is_in_transaction(&self) -> bool

Returns true if the connection is currently in a transaction.

fn transaction<'a, F, R, E>( &'a mut self, callback: F, ) -> impl Future<Output = Result<R, E>> + Send + 'a

where F: for<'c> FnOnce(&'c mut Transaction<'_, Self::Database>) -> Pin<Box<dyn Future<Output = Result<R, E>> + Send + 'c>> + 'a + for<'c> Send + for<'c> Sync, Self: Sized, R: Send, E: From<Error> + Send,

Execute the function inside a transaction.

fn connect( url: &str, ) -> impl Future<Output = Result<Self, Error>> + Send + 'static

where Self: Sized,

Establish a new database connection.

fn connect_with( options: &Self::Options, ) -> impl Future<Output = Result<Self, Error>> + Send

where Self: Sized,

Establish a new database connection with the provided options.

impl Debug for DoConnection

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'c> Executor<'c> for &'c DoConnection

fn execute<'e, 'q: 'e, E>( self, query: E, ) -> BoxFuture<'e, Result<DoQueryResult, Error>>

where E: 'q + Execute<'q, Do>, 'c: 'e,

Reports rows_affected and last_insert_rowid but no rows.

fn fetch_many<'e, 'q: 'e, E>( self, query: E, ) -> BoxStream<'e, Result<Either<DoQueryResult, DoRow>, Error>>

where E: 'q + Execute<'q, Do>, 'c: 'e,

Yields the rows only, never a DoQueryResult. Use execute for what a statement changed.

type Database = Do

fn execute_many<'e, 'q: 'e, E>( self, query: E, ) -> BoxStream<'e, Result<DoQueryResult, Error>>

where E: 'q + Execute<'q, Do>, 'c: 'e,

Execute multiple queries and return the rows affected from each query, in a stream.

fn fetch_optional<'e, 'q: 'e, E>( self, query: E, ) -> BoxFuture<'e, Result<Option<DoRow>, Error>>

where E: 'q + Execute<'q, Do>, 'c: 'e,

Execute the query and returns at most one row.

fn prepare_with<'e>( self, sql: SqlStr, _parameters: &'e [DoTypeInfo], ) -> BoxFuture<'e, Result<DoStatement, Error>>

where 'c: 'e,

Prepare the SQL query, with parameter type information, to inspect the type information about its parameters and results.

fn fetch<'e, 'q, E>( self, query: E, ) -> Pin<Box<dyn Stream<Item = Result<<Self::Database as Database>::Row, Error>> + Send + 'e>>

where 'q: 'e, 'c: 'e, E: 'q + Execute<'q, Self::Database>,

Execute the query and return the generated results as a stream.

fn fetch_all<'e, 'q, E>( self, query: E, ) -> Pin<Box<dyn Future<Output = Result<Vec<<Self::Database as Database>::Row>, Error>> + Send + 'e>>

where 'q: 'e, 'c: 'e, E: 'q + Execute<'q, Self::Database>,

Execute the query and return all the generated results, collected into a Vec.

fn fetch_one<'e, 'q, E>( self, query: E, ) -> Pin<Box<dyn Future<Output = Result<<Self::Database as Database>::Row, Error>> + Send + 'e>>

where 'q: 'e, 'c: 'e, E: 'q + Execute<'q, Self::Database>,

Execute the query and returns exactly one row.

fn prepare<'e>( self, query: SqlStr, ) -> Pin<Box<dyn Future<Output = Result<<Self::Database as Database>::Statement, Error>> + Send + 'e>>

where 'c: 'e,

Prepare the SQL query to inspect the type information of its parameters and results.

impl<'c> Executor<'c> for &'c mut DoConnection

type Database = Do

fn execute<'e, 'q: 'e, E>( self, query: E, ) -> BoxFuture<'e, Result<DoQueryResult, Error>>

where E: 'q + Execute<'q, Do>, 'c: 'e,

Execute the query and return the total number of rows affected.

fn execute_many<'e, 'q: 'e, E>( self, query: E, ) -> BoxStream<'e, Result<DoQueryResult, Error>>

where E: 'q + Execute<'q, Do>, 'c: 'e,

Execute multiple queries and return the rows affected from each query, in a stream.

fn fetch_many<'e, 'q: 'e, E>( self, query: E, ) -> BoxStream<'e, Result<Either<DoQueryResult, DoRow>, Error>>

where E: 'q + Execute<'q, Do>, 'c: 'e,

Execute multiple queries and return the generated results as a stream from each query, in a stream.

fn fetch_optional<'e, 'q: 'e, E>( self, query: E, ) -> BoxFuture<'e, Result<Option<DoRow>, Error>>

where E: 'q + Execute<'q, Do>, 'c: 'e,

Execute the query and returns at most one row.

fn prepare_with<'e>( self, sql: SqlStr, parameters: &'e [DoTypeInfo], ) -> BoxFuture<'e, Result<DoStatement, Error>>

where 'c: 'e,

Prepare the SQL query, with parameter type information, to inspect the type information about its parameters and results.

fn fetch<'e, 'q, E>( self, query: E, ) -> Pin<Box<dyn Stream<Item = Result<<Self::Database as Database>::Row, Error>> + Send + 'e>>

where 'q: 'e, 'c: 'e, E: 'q + Execute<'q, Self::Database>,

Execute the query and return the generated results as a stream.

fn fetch_all<'e, 'q, E>( self, query: E, ) -> Pin<Box<dyn Future<Output = Result<Vec<<Self::Database as Database>::Row>, Error>> + Send + 'e>>

where 'q: 'e, 'c: 'e, E: 'q + Execute<'q, Self::Database>,

Execute the query and return all the generated results, collected into a Vec.

fn fetch_one<'e, 'q, E>( self, query: E, ) -> Pin<Box<dyn Future<Output = Result<<Self::Database as Database>::Row, Error>> + Send + 'e>>

where 'q: 'e, 'c: 'e, E: 'q + Execute<'q, Self::Database>,

Execute the query and returns exactly one row.

fn prepare<'e>( self, query: SqlStr, ) -> Pin<Box<dyn Future<Output = Result<<Self::Database as Database>::Statement, Error>> + Send + 'e>>

where 'c: 'e,

Prepare the SQL query to inspect the type information of its parameters and results.

impl From<Storage> for DoConnection

fn from(storage: Storage) -> Self

Converts to this type from the input type.

impl Send for DoConnection

impl Sync for DoConnection