Databases with sqlx

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production.

The reference material consulted while preparing these pages is listed in the section bibliography.

This page is written against sqlx 0.9.0 with axum 0.8.9 on Rust 1.99 (edition 2024) and PostgreSQL 16, checked on 2026-10-06. sqlx is an async SQL toolkit, not an ORM: you write SQL, and sqlx checks it against a real database at compile time, binds parameters safely and maps rows to Rust types. SQL itself — joins, indexes, transactions, query plans — is covered in this site’s SQL Reference; ORMs are compared on SeaORM & Diesel, and migrations and transactions on the next page.

[dependencies]
sqlx = { version = "0.9.0", features = ["runtime-tokio", "tls-rustls", "postgres", "macros", "migrate", "time", "uuid"] }

Source: sqlx 0.9.0: feature flags

sqlx 0.9 requires Rust 1.94 or later. Since 0.9 the project lives in the transact-rs GitHub organisation (it was launchbadge/sqlx), and the combined runtime-and-TLS features such as runtime-tokio-rustls were removed: select runtime-tokio and tls-rustls (or tls-native-tls) separately.

Where it runs in the request

In the request lifecycle (tokio, hyper, tower layers, router, extractors, handler, response), database access belongs to the handler phase and the services it calls:

  • The PgPool is created once in main, before axum::serve, and reaches handlers through State — an extractor that only clones a handle, so no connection is taken while extractors run.

  • A connection is checked out of the pool when the handler runs its first query (fetch_one(&pool)), and returned as soon as that query finishes; a transaction keeps one connection until commit, rollback or drop. Handlers that are waiting for a connection count against your timeouts (TimeoutLayer) like any other await.

  • A streamed response (fetch turned into a body stream) keeps its connection after the handler has returned, until the last row has been written to the client — slow clients then hold pool connections.

  • Database errors travel back as values: the handler maps sqlx::Error into the application’s error type, whose IntoResponse turns it into a status code on the way out (see Error Handling).

The Bookshelf schema

The examples on this and the next pages use this schema, created by a migration:

CREATE TABLE authors (
    id   BIGSERIAL PRIMARY KEY,
    name TEXT NOT NULL
);

CREATE TABLE books (
    isbn      TEXT PRIMARY KEY CHECK (isbn ~ '^[0-9]{13}$'),
    title     TEXT NOT NULL,
    author_id BIGINT NOT NULL REFERENCES authors (id),
    year      SMALLINT NOT NULL,
    copies    INT NOT NULL DEFAULT 1 CHECK (copies >= 0),
    version   BIGINT NOT NULL DEFAULT 1
);

CREATE TABLE reviews (
    id         BIGSERIAL PRIMARY KEY,
    isbn       TEXT NOT NULL REFERENCES books (isbn) ON DELETE CASCADE,
    rating     SMALLINT NOT NULL CHECK (rating BETWEEN 1 AND 5),
    body       TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX reviews_isbn_idx ON reviews (isbn);

Source: PostgreSQL 16: CREATE TABLE

A pool in state and checked queries

use std::time::Duration;

use axum::{
    Json, Router,
    extract::{Path, State},
    http::StatusCode,
    routing::get,
};
use serde::Serialize;
use sqlx::postgres::{PgPool, PgPoolOptions};

#[derive(Serialize)]
struct BookDetails {
    isbn: String,
    title: String,
    author: String,
    year: i16,
    average_rating: Option<f64>,
}

async fn show_book(State(pool): State<PgPool>, Path(isbn): Path<String>) -> Result<Json<BookDetails>, StatusCode> {
    // checked at compile time against DATABASE_URL: columns, types and nullability
    let book = sqlx::query_as!(
        BookDetails,
        r#"SELECT b.isbn, b.title, a.name AS author, b.year,
                  (SELECT avg(r.rating)::float8 FROM reviews r WHERE r.isbn = b.isbn) AS average_rating
           FROM books b JOIN authors a ON a.id = b.author_id
           WHERE b.isbn = $1"#,
        isbn
    )
    .fetch_optional(&pool)
    .await
    .map_err(|err| {
        tracing::error!(%err, "query failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    book.map(Json).ok_or(StatusCode::NOT_FOUND)
}

async fn count_books(State(pool): State<PgPool>) -> Result<String, StatusCode> {
    let n: i64 = sqlx::query_scalar!(r#"SELECT count(*) AS "n!" FROM books"#)
        .fetch_one(&pool)
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
    Ok(format!("{n} books\n"))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt().with_max_level(tracing::Level::INFO).init();
    let url = std::env::var("DATABASE_URL")?; // e.g. postgres://bookshelf:<password>@localhost/bookshelf
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .min_connections(1)
        .acquire_timeout(Duration::from_secs(3)) // fail fast instead of queueing forever
        .idle_timeout(Duration::from_secs(600))
        .connect(&url)
        .await?;

    let app = Router::new()
        .route("/api/v1/books/{isbn}", get(show_book))
        .route("/api/v1/stats/books", get(count_books))
        .with_state(pool);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

Source: sqlx: query_as! and axum example: sqlx-postgres

$ curl -s localhost:3000/api/v1/books/9780441013593
{"isbn":"9780441013593","title":"Dune","author":"Frank Herbert","year":1965,"average_rating":5.0}
$ curl -s localhost:3000/api/v1/stats/books
3 books

Source: sqlx: PgPoolOptions

  • PgPool is a cheap-to-clone handle (an Arc inside), so it goes into state directly or into an AppState with FromRef (see State & Dependency Injection).

  • max_connections caps concurrent queries per instance. The total across all replicas must stay below the database’s max_connections; size it from measurements, starting around 2-4 times the CPU cores of the database divided by the number of instances. acquire_timeout turns pool exhaustion into a fast error (sqlx::Error::PoolTimedOut, a 503 in Bookshelf's AppError) instead of an ever-growing queue.

  • connect opens one connection eagerly so misconfiguration fails at start-up; connect_lazy defers it.

What the macros check

query!, query_as! and query_scalar! send the SQL to the database named by DATABASE_URL at compile time, ask it to prepare the statement, and generate code with the exact column types. A typo is a compile error:

error: error returned from database: column b.titel does not exist at line 3729
  --> src/routes/books.rs:24:16
   |
24 |     let book = sqlx::query_as!(
   |                ^^^^^^^^^^^^^^^^

Source: sqlx: query! requirements

  • Nullability is inferred: average_rating is Option<f64> because avg can be NULL. Override with column aliases: AS "n!" forces non-null, AS "n?" forces nullable, AS "n: Isbn" sets a type.

  • query_as! maps by column name into any struct with matching fields; it does not need FromRow.

  • Parameters ($1) are always bound, never interpolated, so the macros are immune to SQL injection.

Offline mode: building without a database

CI and Docker builds usually have no database. cargo sqlx prepare (from sqlx-cli) runs the macros once against a live database and stores their metadata in a .sqlx directory, which you commit. With SQLX_OFFLINE=true (or simply no DATABASE_URL), the macros read .sqlx instead:

$ cargo install sqlx-cli --no-default-features --features postgres,rustls
$ export DATABASE_URL=postgres://bookshelf:<password>@localhost/bookshelf
$ cargo sqlx prepare -- --all-targets   # writes .sqlx/query-<hash>.json for every macro invocation
$ git add .sqlx
$ cargo sqlx prepare --check -- --all-targets  # in CI: fail if .sqlx is stale

Source: sqlx: Offline mode

sqlx 0.9 also reads an optional per-crate sqlx.toml (feature sqlx-toml) to rename DATABASE_URL, set global type overrides or relocate the migrations table — useful in workspaces with several databases.

Runtime queries, FromRow and SqlSafeStr

When the SQL is only known at runtime — user-selected sort columns, optional filters — use the functions sqlx::query, query_as and query_scalar (no !), mapping rows with #[derive(sqlx::FromRow)]. Since sqlx 0.9 these functions take impl SqlSafeStr, which is implemented only for &'static str and for strings wrapped in AssertSqlSafe. A String built with format! no longer compiles by accident; you must assert that you built it safely:

use axum::{
    Json, Router,
    extract::{Query, State},
    http::StatusCode,
    routing::get,
};
use serde::{Deserialize, Serialize};
use sqlx::{AssertSqlSafe, FromRow, PgPool, Postgres, QueryBuilder};

#[derive(Serialize, FromRow)]
struct BookRow {
    isbn: String,
    title: String,
    year: i16,
}

#[derive(Deserialize)]
struct Filters {
    sort: Option<String>,
    min_year: Option<i16>,
    q: Option<String>,
}

async fn list(State(pool): State<PgPool>, Query(f): Query<Filters>) -> Result<Json<Vec<BookRow>>, StatusCode> {
    // 1. dynamic ORDER BY: never interpolate user input; map it onto an allow-list
    let order = match f.sort.as_deref() {
        Some("year") => "year, isbn",
        Some("title") => "title, isbn",
        _ => "isbn",
    };

    // 2. optional filters: QueryBuilder binds every value
    let mut qb: QueryBuilder<Postgres> = QueryBuilder::new("SELECT isbn, title, year FROM books WHERE true");
    if let Some(min) = f.min_year {
        qb.push(" AND year >= ").push_bind(min);
    }
    if let Some(q) = f.q {
        qb.push(" AND title ILIKE ").push_bind(format!("%{q}%"));
    }
    qb.push(" ORDER BY ").push(order).push(" LIMIT 100");

    let books = qb.build_query_as::<BookRow>().fetch_all(&pool).await.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
    Ok(Json(books))
}

async fn count_since(State(pool): State<PgPool>) -> Result<String, StatusCode> {
    // a runtime-built string must be asserted safe explicitly (sqlx 0.9)
    let table = "books"; // from code, not from the request
    let sql = format!("SELECT count(*) FROM {table} WHERE year >= $1");
    let n: i64 = sqlx::query_scalar(AssertSqlSafe(sql))
        .bind(1965_i16)
        .fetch_one(&pool)
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
    Ok(format!("{n}\n"))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    let app = Router::new()
        .route("/api/v1/books", get(list))
        .route("/api/v1/stats/since-1965", get(count_since))
        .with_state(pool);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

Source: sqlx: SqlSafeStr and QueryBuilder

$ curl -s 'localhost:3000/api/v1/books?min_year=1969&sort=title'
[{"isbn":"9780593099322","title":"Dune Messiah","year":1969},{"isbn":"9780441478125","title":"The Left Hand of Darkness","year":1969}]

Source: sqlx: QueryBuilder::push_bind

Wrapping attacker-controlled text in AssertSqlSafe defeats the purpose; the wrapper is a reviewable marker, like unsafe. Typical mappings: SMALLINT → i16, INT → i32, BIGINT/BIGSERIAL → i64, TEXT → String, TIMESTAMPTZ → time::OffsetDateTime (feature time) or chrono::DateTime<Utc>, UUID → uuid::Uuid, JSONB → sqlx::types::Json<T>. Newtypes such as Isbn can derive sqlx::Type with #[sqlx(transparent)].

Streaming rows

fetch_all loads every row into a Vec. For exports, fetch returns a Stream that yields rows as Postgres sends them. That stream borrows its executor, so it cannot be returned from the handler directly; a spawned task that owns the pool drives it and feeds Body::from_stream through a bounded channel, keeping memory flat:

use axum::{Router, body::Body, extract::State, http::header, response::IntoResponse, routing::get};
use futures::StreamExt;
use sqlx::PgPool;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

async fn export(State(pool): State<PgPool>) -> impl IntoResponse {
    // The row stream borrows the pool, so it is driven by a task that owns the pool and
    // forwards lines through a bounded channel (which also applies back-pressure).
    let (tx, rx) = mpsc::channel::<Result<String, std::io::Error>>(32);
    tokio::spawn(async move {
        let mut rows = sqlx::query_as::<_, (String, String)>("SELECT isbn, title FROM books ORDER BY isbn").fetch(&pool);
        while let Some(row) = rows.next().await {
            let line = row.map(|(isbn, title)| format!("{isbn},{title}\n")).map_err(std::io::Error::other);
            let failed = line.is_err();
            if tx.send(line).await.is_err() || failed {
                break; // client went away, or the query failed (the response is truncated)
            }
        }
    });
    ([(header::CONTENT_TYPE, "text/csv")], Body::from_stream(ReceiverStream::new(rx)))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    let app = Router::new().route("/exports/books.csv", get(export)).with_state(pool);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

Source: sqlx: fetch

The stream holds a pool connection until it finishes, so long exports reduce the connections available to other requests; give exports their own small pool or a read replica.

The repository pattern

Bookshelf's services depend on traits, and the db module implements them with sqlx (see Project Structure). Repositories own the SQL, return domain types and map database errors in one place:

use sqlx::PgPool;

#[derive(Debug)]
pub struct Book {
    pub isbn: String,
    pub title: String,
    pub copies: i32,
}

#[derive(Clone)]
pub struct PgBookRepo {
    pool: PgPool,
}

impl PgBookRepo {
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }

    #[tracing::instrument(skip(self), fields(db.system = "postgresql", db.operation = "SELECT"))]
    pub async fn find(&self, isbn: &str) -> sqlx::Result<Option<Book>> {
        sqlx::query_as!(Book, "SELECT isbn, title, copies FROM books WHERE isbn = $1", isbn)
            .fetch_optional(&self.pool)
            .await
    }

    /// Several books in ONE query, instead of one query per ISBN (N+1).
    pub async fn find_many(&self, isbns: &[String]) -> sqlx::Result<Vec<Book>> {
        sqlx::query_as!(Book, "SELECT isbn, title, copies FROM books WHERE isbn = ANY($1) ORDER BY isbn", isbns)
            .fetch_all(&self.pool)
            .await
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt().with_max_level(tracing::Level::DEBUG).init();
    let repo = PgBookRepo::new(PgPool::connect(&std::env::var("DATABASE_URL")?).await?);
    println!("{:?}", repo.find("9780441013593").await?);
    let wanted = vec!["9780441013593".to_owned(), "9780593099322".to_owned()];
    println!("{} books in one round trip", repo.find_many(&wanted).await?.len());
    Ok(())
}

Source: sqlx: query_as!

Avoiding N+1 queries

The N+1 problem is one query for a list plus one query per item — 1 + 50 round trips for a page of 50 books with their authors. With hand-written SQL the fixes are explicit: a JOIN (as in show_book above), a second query with = ANY($1) for all ids at once (as in find_many), or a json_agg subquery that returns the children as JSON. Watch the query count per request in traces; a page that issues dozens of queries is the most common cause of slow endpoints. See Diagnosing and Fixing Slow Queries for reading query plans.

Transaction scope and lock contention

Gangappa & Rangaraj (chapter 11) trace many "the database is slow" incidents back to the application holding database resources too long, not to slow SQL. Two resources matter: the pooled connection a transaction occupies, and the row locks it holds.

  • Keep transactions short. A transaction pins one connection from begin until commit or rollback. With max_connections = 10, ten requests that each keep a transaction open while waiting 2 seconds for an HTTP call starve every other request: they queue in acquire until acquire_timeout turns them into 503 errors, although Postgres itself is idle. Never .await external I/O — HTTP calls, e-mail, a Redis round trip, a channel to another task — between begin and commit. Read what you need, commit, then do the I/O; if the I/O must be atomic with the write, record the intent in the same transaction (an outbox row) and let a background job perform it.

  • Locks are held until commit. Every UPDATE, DELETE and SELECT ... FOR UPDATE takes row locks that block other writers of the same rows until the transaction ends, so a long transaction also slows other requests touching the same rows. Hot rows — a per-book counter updated on every loan, a single "next invoice number" row — serialise all their writers; spread them (one row per day or per shard, summed on read) or use a sequence.

  • Bound the wait. SET LOCAL lock_timeout = '2s' inside a transaction makes a blocked statement fail with an error (SQLSTATE 55P03) instead of waiting indefinitely; statement_timeout does the same for slow statements.

  • Lock in a consistent order (for example by primary key) when a transaction updates several rows; two transactions locking the same rows in opposite orders deadlock, and Postgres aborts one of them (SQLSTATE 40P01).

For work queues, FOR UPDATE SKIP LOCKED turns lock contention into parallelism: each worker claims the rows nobody else has locked, instead of queueing behind them. Bookshelf's loan reminders use a small table:

CREATE TABLE reminders (
    id      BIGSERIAL PRIMARY KEY,
    isbn    TEXT NOT NULL REFERENCES books (isbn),
    due_at  TIMESTAMPTZ NOT NULL,
    sent_at TIMESTAMPTZ
);
CREATE INDEX reminders_due_idx ON reminders (due_at) WHERE sent_at IS NULL;

Source: PostgreSQL 16: The Locking Clause

use sqlx::PgPool;

/// Claims up to `batch` due reminders. Rows locked by another worker are skipped instead of waited
/// for, so several replicas can share the queue without blocking one another.
async fn claim_due(pool: &PgPool, batch: i64) -> sqlx::Result<Vec<(i64, String)>> {
    let mut tx = pool.begin().await?;
    sqlx::query("SET LOCAL lock_timeout = '2s'").execute(&mut *tx).await?; // never wait forever on a lock
    let claimed: Vec<(i64, String)> = sqlx::query_as(
        "SELECT id, isbn FROM reminders
         WHERE sent_at IS NULL AND due_at <= now()
         ORDER BY due_at
         LIMIT $1
         FOR UPDATE SKIP LOCKED",
    )
    .bind(batch)
    .fetch_all(&mut *tx)
    .await?;
    let ids: Vec<i64> = claimed.iter().map(|(id, _)| *id).collect();
    sqlx::query("UPDATE reminders SET sent_at = now() WHERE id = ANY($1)").bind(&ids).execute(&mut *tx).await?;
    tx.commit().await?; // the row locks are held only for these three statements
    Ok(claimed)
}

async fn send_email(isbn: &str) {
    // external I/O happens AFTER commit, never while the transaction holds locks and a pooled connection
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    let _ = isbn;
}

async fn worker(name: &'static str, pool: PgPool) -> sqlx::Result<()> {
    loop {
        let batch = claim_due(&pool, 2).await?;
        if batch.is_empty() {
            return Ok(());
        }
        for (id, isbn) in batch {
            send_email(&isbn).await;
            println!("{name} sent reminder {id} ({isbn})");
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    sqlx::query("INSERT INTO reminders (isbn, due_at) SELECT '9780441013593', now() FROM generate_series(1, 5)")
        .execute(&pool)
        .await?;
    let (a, b) = tokio::join!(worker("worker-a", pool.clone()), worker("worker-b", pool.clone()));
    a?;
    b?;
    Ok(())
}

Source: sqlx: Transaction and PostgreSQL 16: lock_timeout

$ cargo run
worker-b sent reminder 1 (9780441013593)
worker-b sent reminder 2 (9780441013593)
worker-b sent reminder 3 (9780441013593)
worker-a sent reminder 5 (9780441013593)
worker-b sent reminder 4 (9780441013593)

Source: PostgreSQL 16: Row-Level Locks

  • The two workers never claim the same reminder, and neither waits for the other’s locks.

  • Marking a reminder as sent before sending gives at-most-once delivery: a crash between commit and send_email loses that reminder. For at-least-once delivery, store a claimed_until lease instead of sent_at and set sent_at after sending; reminders whose lease expired are claimed again.

  • To see contention in production, look for sessions with wait_event_type = 'Lock' in pg_stat_activity (and the blocking process with pg_blocking_pids(pid)), and enable log_lock_waits so waits longer than deadlock_timeout are logged. The application-side signal is the pool’s acquire time (next section).

Optimistic locking with a version column, isolation levels and SELECT ... FOR UPDATE in request handlers are on the Migrations & Transactions page.

Instrumentation and pool metrics

Gangappa & Rangaraj (Mastering Distributed Observability in Rust, chapter 5) recommend a span per database call with the OpenTelemetry semantic-convention attributes — db.system, db.operation, db.collection.name (the table) — and sanitised statements: record the SQL text with placeholders, never the bound values, which may contain personal data. sqlx itself logs each statement at DEBUG (target sqlx::query, with elapsed time) and slow statements at WARN (configurable with ConnectOptions::log_slow_statements).

Pool saturation is the early warning of database trouble (chapter 11): export pool.size() and pool.num_idle() as gauges and the time spent in acquire as a histogram, and alert when idle connections stay at zero (see Metrics & SLOs).

Other databases

  • tokio-postgres with a bb8 or deadpool pool is a lower-level Postgres driver without compile-time checks; the official tokio-postgres example shows it in state.

  • MySQL/MariaDB and SQLite are supported by sqlx with the same API (features mysql, sqlite).

  • MongoDB has an official async Rust driver (mongodb crate, also an official axum example); see MongoDB Drivers & Tooling.

Where the books are outdated

Gangappa & Rangaraj use sqlx 0.8, where sqlx::query(&format!(...)) compiled; in 0.9 it requires AssertSqlSafe, and the MSRV is 1.94. The 0.8 runtime-tokio-rustls feature is gone (use runtime-tokio plus tls-rustls). Yoon (Beginning Axum, chapter 3) uses SeaORM rather than sqlx; see SeaORM & Diesel.

References