sqlx

sqlx — Rust 异步 SQL 工具包

sqlx 是一款以异步为先的纯 Rust SQL 工具包，支持编译时查询检查。它支持 PostgreSQL、MySQL 和 SQLite，且无需 DSL——你可以直接编写原始 SQL，sqlx 会在编译时针对实时数据库对其进行验证。

Crate 架构

模块	用途
sqlx::postgres	PostgreSQL 驱动（PgPool、PgConnection、PgListener）
sqlx::mysql	MySQL 驱动（MySqlPool、MySqlConnection）
sqlx::sqlite	SQLite 驱动（SqlitePool、SqliteConnection）
sqlx::any	与数据库无关的驱动（AnyPool）——不进行编译时检查
sqlx::query_builder	运行时动态构建查询（QueryBuilder）
sqlx::migrate	迁移框架（migrate! 宏、Migrator）
sqlx::types	类型包装器（Json<T>、Text<T>）

快速入门：最小依赖配置

务必至少包含一个数据库驱动和一个运行时功能。若未启用运行时功能，连接池会在运行时触发 panic。

# Cargo.toml
[dependencies]
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "chrono", "uuid"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
dotenvy = "0.15"

# .env
DATABASE_URL=postgres://postgres:password@localhost:5432/mydb

use sqlx::PgPool;

#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
    dotenvy::dotenv().ok();
    let pool = PgPool::connect(&std::env::var("DATABASE_URL").unwrap()).await?;
    let users = sqlx::query_as!(User, "SELECT id, name FROM users")
        .fetch_all(&pool).await?;
    Ok(())
}

重要：必须在编译时设置 DATABASE_URL，query!() 宏才能正常工作。请在 .env 中设置，或使用 SQLX_OFFLINE=true 和预先准备的元数据（参见“离线模式”部分）。

快速参考：我应该阅读哪篇指南？

• 编译时查询宏、类型覆盖、离线模式 -> 阅读 references/compile-time-checking.md
• 连接池、PgConnectOptions、连接池调优 -> 阅读 references/connection-pooling.md
• Rust 到 SQL 的类型映射、FromRow、自定义类型 -> 阅读 references/type-mapping.md
• 事务、嵌套事务、保存点 -> 阅读 references/transactions-migrations.md
• #[sqlx::test]、测试数据、测试模式 -> 阅读 references/testing-mocking.md
• PostgreSQL 特性（PgListener、COPY、数组、范围） -> 阅读 references/database-specifics.md
• 错误处理、sqlx::Error 枚举 -> 阅读 references/error-handling.md
• 功能标志、Cargo.toml 配置模式 -> 阅读 references/feature-flags.md

核心模式一览

1. 使用编译时检查进行查询（推荐）

query!() 宏会在编译时连接数据库，验证 SQL 语法，检查参数类型是否匹配，并验证返回列的类型。

// Returns an anonymous struct with typed fields (no FromRow needed)
let record = sqlx::query!("SELECT id, name, email FROM users WHERE id = $1", user_id)
    .fetch_one(&pool)
    .await?;
// record.id: i64, record.name: String, record.email: String

// query_as! maps to a named struct (needs #[derive(FromRow)])
let user: User = sqlx::query_as!(User, "SELECT id, name FROM users WHERE id = $1", user_id)
    .fetch_one(&pool)
    .await?;

// query_scalar! returns a single column value
let count: i64 = sqlx::query_scalar!("SELECT COUNT(*)::BIGINT FROM users")
    .fetch_one(&pool)
    .await?;

2. 运行时查询函数（无需编译时数据库）

无法在编译时设置 DATABASE_URL 时，请使用这些函数。它们不会进行类型检查。

// query() returns anonymous PgRow — access columns by name or index
// Note: `sqlx::Row` trait must be in scope for .get()
use sqlx::Row;

let row = sqlx::query("SELECT id, name FROM users WHERE id = $1")
    .bind(1)
    .fetch_one(&pool)
    .await?;
let id: i64 = row.get("id");
let name: String = row.get("name");

3. 获取策略

每个查询都支持多种获取方法。请根据预期的结果数量进行选择：

// .fetch_one()       — Exactly one row. Error if 0 rows (RowNotFound) or >1 rows
// .fetch_optional()  — 0 or 1 row. Returns Option<T>. None if no rows.
// .fetch_all()       — All rows as Vec<T>. Empty vec if no rows.
// .fetch(_)          — Returns a Stream (row by row, lazy). Use with futures::StreamExt.

4. 连接池配置

use sqlx::postgres::PgPoolOptions;
use std::time::Duration;

let pool = PgPoolOptions::new()
    .max_connections(20)
    .min_connections(5)
    .acquire_timeout(Duration::from_secs(3))
    .idle_timeout(Duration::from_secs(600))
    .max_lifetime(Duration::from_secs(1800))
    .connect("postgres://user:pass@localhost/db")
    .await?;

5. 事务

事务实现了 Executor，因此你可以将其传递给任何接受连接池或连接的函数。

use sqlx::Acquire;

// Manual transaction
let mut tx = pool.begin().await?;
sqlx::query!("INSERT INTO users (name) VALUES ($1)", "Alice")
    .execute(&mut *tx)
    .await?;
sqlx::query!("INSERT INTO audit_log (action) VALUES ($1)", "user created")
    .execute(&mut *tx)
    .await?;
tx.commit().await?;
// Dropping tx without commit() triggers automatic rollback.

// Closure-based (auto-commit on Ok, auto-rollback on Err)
// Requires acquiring a connection first — .transaction() is on Connection, not Pool.
let mut conn = pool.acquire().await?;
let result = conn.transaction::<_, _, sqlx::Error>(|tx| {
    Box::pin(async move {
        sqlx::query!("INSERT INTO users (name) VALUES ($1)", "Bob")
            .execute(&mut **tx).await?;
        Ok(())
    })
}).await?;

// Nested transactions (SAVEPOINTs)
let mut tx = pool.begin().await?;
let mut nested = tx.begin().await?; // Creates a SAVEPOINT — requires `Acquire`
nested.rollback().await?;           // Only rolls back to savepoint
tx.commit().await?;                 // Outer transaction unaffected

6. 使用 Executor 编写泛型函数

使用 Executor trait 编写可同时适用于连接池、连接和事务的函数：

use sqlx::{Executor, PgExecutor};

async fn find_user_by_id<'e, E>(executor: E, id: i64) -> sqlx::Result<User>
where
    E: PgExecutor<'e>,
{
    sqlx::query_as!(User, "SELECT * FROM users WHERE id = $1", id)
        .fetch_one(executor)
        .await
}

// Works with pool, connection, or transaction:
let user = find_user_by_id(&pool, 1).await?;
let user = find_user_by_id(&mut tx, 1).await?;

7. FromRow 派生

使用 #[derive(FromRow)] 将 SQL 行映射到 Rust 结构体。结构体字段名必须与 SQL 列名匹配。

use sqlx::FromRow;

#[derive(Debug, FromRow)]
struct User {
    id: i64,
    name: String,
    email: String,
    created_at: chrono::DateTime<chrono::Utc>,
}

// Column renaming when names don't match
#[derive(Debug, FromRow)]
struct User {
    id: i64,
    #[sqlx(rename = "full_name")]
    name: String,
}

// Flatten for JOINs (combines fields from multiple tables)
#[derive(Debug, FromRow)]
struct UserWithPost {
    #[sqlx(flatten)]
    user: User,
    #[sqlx(flatten)]
    post: Post,
}

// Skip columns from the query that you don't need
#[derive(Debug, FromRow)]
struct UserSummary {
    id: i64,
    name: String,
    #[sqlx(skip)]
    _ignored_field: String,
}

8. JSON 列

use sqlx::types::Json;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
struct User {
    id: i64,
    #[sqlx(json)]
    metadata: Json<Metadata>,
}

#[derive(Debug, Serialize, Deserialize)]
struct Metadata {
    role: String,
    preferences: std::collections::HashMap<String, String>,
}

// Insert with JSON value
sqlx::query!(
    "INSERT INTO users (name, metadata) VALUES ($1, $2)",
    "Alice",
    Json(Metadata { role: "admin".into(), preferences: Default::default() })
).execute(&pool).await?;

// Read — Json<T> automatically deserializes
let user: User = sqlx::query_as(
    "SELECT id, name, metadata FROM users WHERE id = $1"
).bind(1).fetch_one(&pool).await?;
println!("Role: {}", user.metadata.0.role);

9. 使用 QueryBuilder 构建动态查询

当 SQL 结构取决于运行时条件时，请使用 QueryBuilder，而不是拼接字符串。

use sqlx::QueryBuilder;

let mut qb = QueryBuilder::new("SELECT * FROM users WHERE ");

// Conditionally add clauses
if let Some(name) = &filter.name {
    qb.push("name LIKE ").push_bind(format!("%{}%", name));
}

if filter.active_only {
    if filter.name.is_some() {
        qb.push(" AND ");
    }
    qb.push("active = ").push_bind(true);
}

let users: Vec<User> = qb.build_query_as::<User>()
    .fetch_all(&pool)
    .await?;

10. 迁移

# Create a new migration
sqlx migrate add create_users

# Run all pending migrations
sqlx migrate run

# Rollback the last migration
sqlx migrate revert

# Show migration status
sqlx migrate info

在代码中嵌入并运行迁移：

let migrator = sqlx::migrate!(); // Reads from ./migrations by default
migrator.run(&pool).await?;

11. #[sqlx::test] 属性

自动创建隔离的测试数据库、运行迁移，并在之后进行清理。

#[sqlx::test(migrations = "./migrations")]
async fn test_create_user(pool: PgPool) -> sqlx::Result<()> {
    sqlx::query!("INSERT INTO users (name) VALUES ($1)", "Alice")
        .execute(&pool).await?;

    let user = sqlx::query_as!(User, "SELECT * FROM users WHERE name = $1", "Alice")
        .fetch_one(&pool).await?;

    assert_eq!(user.name, "Alice");
    Ok(())
}

各数据库的占位符语法

这是常见的错误来源。不同的数据库使用不同的占位符：

数据库	占位符	示例
PostgreSQL	$1、$2、$3...	WHERE id = $1 AND name = $2
MySQL	?	WHERE id = ? AND name = ?
SQLite	?	WHERE id = ? AND name = ?

编译时宏会根据 DATABASE_URL 自动处理此问题。运行时查询必须使用目标数据库对应的语法。

关键导入参考

use sqlx::{
    // Query functions (runtime, no compile-time checking)
    query, query_as, query_scalar,
    // Traits
    Executor, FromRow, Decode, Encode, Type,
    // Pool types
    PgPool, MySqlPool, SqlitePool, AnyPool, Pool,
    // Connection types
    PgConnectOptions, Transaction,
    // Error types
    Error, Result,
    // JSON support
    types::Json,
    // PostgreSQL extras
    postgres::{PgConnectOptions, PgListener, PgPoolOptions, PgSslMode},
    // MySQL extras
    mysql::{MySqlConnectOptions, MySqlPoolOptions},
    // SQLite extras
    sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteJournalMode},
};

常见问题

1. 编译时缺少 DATABASE_URL — query!() 宏会在 cargo build 期间连接数据库。请在 .env 中设置 DATABASE_URL，或使用 cargo sqlx prepare 以离线模式运行。

2. 缺少运行时功能标志 — 始终在功能中包含 runtime-tokio 或 runtime-async-std。否则，创建连接池时会发生 panic。

3. 缺少数据库功能标志 — 仅添加 sqlx = "0.8" 而不指定数据库功能，只会启用 any、json、macros、migrate 和 derive。请务必添加例如 features = ["postgres"] 的配置。

4. 占位符语法错误 — PostgreSQL 使用 $1，MySQL/SQLite 使用 ?。编译时宏会强制检查这一点；运行时查询则不会。

5. 事务解引用 — 使用 &mut *tx（而不是 &tx）在事务上执行查询。

6. SQLite 布尔值映射 — 布尔值以 INTEGER（0/1）存储。在 SQL 比较中使用整数。

7. 查询较多时编译速度慢 — 使用离线模式，或考虑对非关键查询使用 query_unchecked!()。

8. RETURNING 子句 — PostgreSQL 支持在 INSERT/UPDATE 中使用 RETURNING *。MySQL 不支持（请改用 LAST_INSERT_ID()）。

参考文件

如需了解任何主题的详细信息，请阅读相应的参考文件：

• references/compile-time-checking.md — query!、query_as!、query_scalar!、类型覆盖、非检查变体、离线模式、.sqlx/ 目录、查询文件
• references/connection-pooling.md — 连接池选项、连接选项、PgConnectOptions、SSL 模式、after_connect 钩子、连接池生命周期
• references/type-mapping.md — 完整的 Rust 到 SQL 类型对照表、功能标志类型、FromRow 属性、自定义 Encode/Decode、透明类型、枚举映射
• references/transactions-migrations.md — 事务（手动、闭包、嵌套）、可逆迁移、migrate! 宏、迁移 CLI 命令
• references/testing-mocking.md — #[sqlx::test] 宏、测试固件、便于模拟的仓储模式、用于测试的内存 SQLite
• references/database-specifics.md — PostgreSQL（PgListener、COPY、数组、范围、复合类型）、MySQL 特性、SQLite（内存、日志模式、PRAGMA、扩展）
• references/error-handling.md — sqlx::Error 枚举、数据库错误代码、RowNotFound、解码错误、连接池错误
• references/feature-flags.md — 完整的功能标志表、适用于不同用例的 Cargo.toml 配置模式、TLS 选项、运行时选项