sqlx-ext

Backend-generic, poll-scoped implicit transaction execution for SQLx.

Features

The default build targets Tokio + PostgreSQL. Production applications can select only the needed SQLx backend/runtime:

# PostgreSQL + Tokio (default)
sqlx-ext = "0.1"

# SQLite + Tokio
sqlx-ext = { version = "0.1", default-features = false, features = ["runtime-tokio", "sqlite"] }

# MySQL + async-std
sqlx-ext = { version = "0.1", default-features = false, features = ["runtime-async-std", "mysql"] }

Available features: runtime-tokio, runtime-async-std, postgres, mysql, and sqlite. Enable exactly one SQLx runtime and the backend(s) your service uses.

use sqlx::Postgres;
use sqlx_ext::{Database, prelude::*};

async fn example() -> Result<(), sqlx_ext::DbError> {
    Database::<Postgres>::init("postgres://localhost/app").await?;
    Database::<Postgres>::transaction_global(|| Box::pin(async {
        sqlx::query::<Postgres>("insert into users (email) values ($1)")
            .bind("alice@example.com")
            .execute_scoped().await?;
        Ok(())
    })).await?;
    Ok(())
}

Initialization choices

Use init() for the usual application-wide setup. Queries outside a transaction then automatically use the registered pool. This is the recommended application entry point: Services and Repositories do not keep a Database, Pool, or Transaction field.

use sqlx::Postgres;
use sqlx_ext::{Database, prelude::*};
async fn example() -> Result<(), sqlx_ext::DbError> {
    Database::<Postgres>::init("postgres://localhost/app").await?;
    let user_count: i64 = sqlx::query_scalar::<Postgres, i64>("SELECT COUNT(*) FROM users")
        .fetch_one_scoped()
        .await?;
    assert!(user_count >= 0);
    Ok(())
}

Recommended layering after initialization:

startup      Database::<Postgres>::init(url).await?
service      Database::<Postgres>::transaction_global(|| ...)
repository   query.execute_scoped().await

For non-transactional work, a Service simply calls its Repository. The scoped query falls back to the pool registered by init(). For transactional work, the Service wraps the same Repository calls in transaction_global; no method signature changes and no executor parameter is passed through the stack.

connect() is an advanced/component-local alternative. It can start a transaction without global registration; only queries outside that transaction need set_global(). Do not inject this handle into ordinary application Services when init() + transaction_global() is sufficient.

use sqlx::Postgres;
use sqlx_ext::{Database, prelude::*};

async fn example() -> Result<(), sqlx_ext::DbError> {
    let db = Database::<Postgres>::connect("postgres://localhost/app").await?;
    db.transaction(|_| Box::pin(async {
        sqlx::query::<Postgres>("UPDATE accounts SET last_seen_at = NOW()")
            .execute_scoped()
            .await?;
        Ok(())
    })).await?;
    Ok(())
}

Transactional services and repositories

After startup initialization, a repository/service receives neither a Pool, Database, nor a Transaction; it simply uses scoped query methods.

use sqlx::Postgres;
use sqlx_ext::{Database, prelude::*};

async fn create_user(email: &str) -> Result<i64, sqlx_ext::DbError> {
    let id: i64 = sqlx::query_scalar::<Postgres, i64>(
        "INSERT INTO users (email) VALUES ($1) RETURNING id",
    )
    .bind(email)
    .fetch_one_scoped()
    .await?;
    Ok(id)
}

async fn example() -> Result<(), sqlx_ext::DbError> {
    Database::<Postgres>::init("postgres://localhost/app").await?;
    Database::<Postgres>::transaction_global(|| Box::pin(async {
        let user_id = create_user("alice@example.test").await?;
        sqlx::query::<Postgres>("INSERT INTO audit_log (user_id, action) VALUES ($1, 'create')")
            .bind(user_id)
            .execute_scoped()
            .await?;
        Ok(())
    })).await?;
    Ok(())
}

The same Repository call outside a transaction needs no wrapper:

async fn create_user(email: &str) -> Result<i64, sqlx_ext::DbError> { Ok(1) }

async fn example() -> Result<(), sqlx_ext::DbError> {
    let user_id = create_user("alice@example.test").await?;
    let _ = user_id;
    Ok(())
}

Returning Err rolls the entire outer transaction back:

use sqlx::Postgres;
use sqlx_ext::{Database, DbError, prelude::*};

async fn example() -> Result<(), DbError> {
    Database::<Postgres>::init("postgres://localhost/app").await?;
    let result = Database::<Postgres>::transaction_global(|| Box::pin(async {
        sqlx::query::<Postgres>("INSERT INTO users (email) VALUES ($1)")
            .bind("will-rollback@example.test")
            .execute_scoped()
            .await?;
        Err::<(), DbError>(DbError::TransactionClosed)
    })).await;
    assert!(result.is_err());
    Ok(())
}

Query API coverage

Import the prelude and use the same suffix for each dynamic SQLx query type.

use sqlx::{Postgres, Row};
use sqlx_ext::{Database, prelude::*};

async fn example() -> Result<(), sqlx_ext::DbError> {
    Database::<Postgres>::init("postgres://localhost/app").await?;
    // Query: execute and row-oriented reads.
    sqlx::query::<Postgres>("DELETE FROM sessions WHERE expires_at < NOW()")
        .execute_scoped().await?;
    let row = sqlx::query::<Postgres>("SELECT id FROM users WHERE email = $1")
        .bind("alice@example.test")
        .fetch_one_scoped().await?;
    let id: i64 = row.try_get("id").unwrap();
    let missing = sqlx::query::<Postgres>("SELECT id FROM users WHERE id = -1")
        .fetch_optional_scoped().await?;
    assert!(missing.is_none());
    let rows = sqlx::query::<Postgres>("SELECT id FROM users LIMIT 100")
        .fetch_all_scoped().await?;

    // QueryAs: map rows to tuples or FromRow types.
    let users: Vec<(i64, String)> = sqlx::query_as::<Postgres, (i64, String)>(
        "SELECT id, email FROM users ORDER BY id LIMIT 100",
    ).fetch_all_scoped().await?;

    // QueryScalar: select a single column.
    let count: i64 = sqlx::query_scalar::<Postgres, i64>("SELECT COUNT(*) FROM users")
        .fetch_one_scoped().await?;
    let _ = (id, rows, users, count);
    Ok(())
}

QueryBuilder is supported after build, build_query_as, or build_query_scalar:

use sqlx::{Postgres, QueryBuilder, Row};
use sqlx_ext::{Database, prelude::*};

async fn example() -> Result<(), sqlx_ext::DbError> {
    Database::<Postgres>::init("postgres://localhost/app").await?;
    let mut builder = QueryBuilder::<Postgres>::new("SELECT ");
    builder.push_bind(7_i32).push(" AS value");
    let row = builder.build().fetch_one_scoped().await?;
    assert_eq!(row.try_get::<i32, _>("value").unwrap(), 7);
    Ok(())
}

For a Stream-shaped consumer API, use fetch_scoped:

use futures::TryStreamExt;
use sqlx::Postgres;
use sqlx_ext::{Database, prelude::*};

async fn example() -> Result<(), sqlx_ext::DbError> {
    Database::<Postgres>::init("postgres://localhost/app").await?;
    let ids: Vec<i64> = sqlx::query_scalar::<Postgres, i64>("SELECT id FROM users")
        .fetch_scoped()
        .try_collect()
        .await?;
    Ok(())
}

Database::<DB>::connect() returns an independent handle and can be used for db.transaction(...) without installing a global pool. Install it with set_global() only when queries also need to run outside a transaction. Most applications should prefer init() and transaction_global() instead.

ScopedQuery, ScopedQueryAs, and ScopedScalarQuery route dynamic SQLx queries to the global pool outside a transaction and the current transaction inside one. The scope is installed only during each Future poll, so it can move between runtime threads. Do not retain Transaction or spawn detached work from a transaction; spawned tasks do not inherit its scope.

The adapters cover every SQLx 0.8 terminal method exposed by the dynamic Query, QueryAs, and QueryScalar objects: execute, execute_many, fetch, fetch_many, fetch_one, fetch_optional, and fetch_all. The scoped method names add the _scoped suffix. SQLx itself marks the *_many methods deprecated because multi-statement prepared queries are SQLite-only; they remain available here for API completeness. fetch_scoped keeps the same Stream-shaped consumption API but currently buffers the complete result before yielding it; this deliberately avoids returning a stream that outlives the mutex guard protecting an implicit transaction.

Nested Database::transaction calls reuse the outer transaction (no savepoint). Consequently, if an inner error is caught by the outer closure, its preceding writes remain part of the outer transaction and will commit if the outer closure returns Ok. Use an uncaught error to roll back the whole transaction.

Transaction must not outlive its closure; retaining it produces DbError::TransactionLeaked and prevents commit. Detached tokio::spawn tasks do not inherit the scope. Cancellation and panic never commit: SQLx drops the still-owned transaction and performs its rollback-on-drop fallback.

For tests or deliberate reconfiguration, call Database::<DB>::reset_global(). Global pools are isolated by SQLx backend type; attempting a query under a scope belonging to another backend returns DbError::BackendScopeMismatch.

The included PostgreSQL integration test runs against DATABASE_URL:

DATABASE_URL=postgres://... cargo test --test postgres_scoped

Repository integration fixture

The test suite keeps a sample UserRepository trait and SQLx implementation under tests/support; it is not part of the library API. Run its real transaction test with:

DATABASE_URL=postgres://... cargo test --test repository_transaction