本站怎么用 sqlx

sqlx v0.8.6 [_rt-tokio,bigdecimal,chrono,json,postgres,runtime-tokio,runtime-tokio-rustls,rust_decimal,sqlx-postgres,time,tls-rustls-ring,tls-rustls-ring-webpki,uuid]

fn build_connect_options(url: &str) -> ConnectOptions {
  let mut opt = ConnectOptions::new(url.to_string());
  opt
    .max_connections(read_env_u32("DB_MAX_CONN", DEFAULT_MAX_CONNECTIONS))
    .min_connections(read_env_u32("DB_MIN_CONN", DEFAULT_MIN_CONNECTIONS))
    .connect_timeout(Duration::from_secs(read_env_secs(
      "DB_CONNECT_TIMEOUT_SECS",
      DEFAULT_CONNECT_TIMEOUT_SECS,
    )))
    .acquire_timeout(Duration::from_secs(read_env_secs(
      "DB_ACQUIRE_TIMEOUT_SECS",
      DEFAULT_ACQUIRE_TIMEOUT_SECS,
    )))
    .idle_timeout(Duration::from_secs(read_env_secs(
      "DB_IDLE_TIMEOUT_SECS",
      DEFAULT_IDLE_TIMEOUT_SECS,
    )))
    // sqlx 默认 INFO 太吵；DEBUG 仍能打开 SQL 语句日志
    .sqlx_logging_level(tracing::log::LevelFilter::Debug);
  opt
}

    pub fn sqlx_pool_options<DB>(self) -> sqlx::pool::PoolOptions<DB>
    where
        DB: sqlx::Database,
    {
        let mut opt = sqlx::pool::PoolOptions::new();
        if let Some(max_connections) = self.max_connections {
            opt = opt.max_connections(max_connections);
        }
        if let Some(min_connections) = self.min_connections {
            opt = opt.min_connections(min_connections);
        }
        if let Some(connect_timeout) = self.connect_timeout {
            opt = opt.acquire_timeout(connect_timeout);
        }
        if let Some(idle_timeout) = self.idle_timeout {
            opt = opt.idle_timeout(Some(idle_timeout));
        }
        if let Some(acquire_timeout) = self.acquire_timeout {
            opt = opt.acquire_timeout(acquire_timeout);
        }
        if let Some(max_lifetime) = self.max_lifetime {
            opt = opt.max_lifetime(Some(max_lifetime));
        }
        opt = opt.test_before_acquire(self.test_before_acquire);
        opt
    }

let opt = build_connect_options(&lab_url("site-pool"));
show("site ", &opt.clone().sqlx_pool_options::<Postgres>());
show("sqlx ", &sea_orm::sqlx::pool::PoolOptions::<Postgres>::new());

let mut only_connect = ConnectOptions::new(lab_url("x"));
only_connect.connect_timeout(Duration::from_secs(1));
show("only connect_timeout(1s)", &only_connect.sqlx_pool_options::<Postgres>());
let mut both = ConnectOptions::new(lab_url("x"));
both.acquire_timeout(Duration::from_secs(5)).connect_timeout(Duration::from_secs(1));
show("acquire_timeout(5s) + connect_timeout(1s)", &both.sqlx_pool_options::<Postgres>());

let db = Database::connect(opt).await?;
let pool = db.get_postgres_connection_pool();
tokio::time::sleep(Duration::from_millis(300)).await;
println!(
  "after Database::connect: size={} num_idle={} backends={:?}",
  pool.size(),
  pool.num_idle(),
  backends(&mut obs, "site-pool").await?
);

site : max=32 min=2 acquire_timeout=5s idle_timeout=Some(600s) max_lifetime=Some(1800s) test_before_acquire=true slow_threshold=2s
sqlx : max=10 min=0 acquire_timeout=30s idle_timeout=Some(600s) max_lifetime=Some(1800s) test_before_acquire=true slow_threshold=2s
only connect_timeout(1s): max=10 min=0 acquire_timeout=1s idle_timeout=Some(600s) max_lifetime=Some(1800s) test_before_acquire=true slow_threshold=2s
acquire_timeout(5s) + connect_timeout(1s): max=10 min=0 acquire_timeout=5s idle_timeout=Some(600s) max_lifetime=Some(1800s) test_before_acquire=true slow_threshold=2s
after Database::connect: size=2 num_idle=2 backends=[(821, "idle"), (822, "idle")]

        let mut sqlx_opts = options
            .url
            .parse::<PgConnectOptions>()
            .map_err(sqlx_error_to_conn_err)?;
        use sqlx::ConnectOptions;
        if !options.sqlx_logging {
            sqlx_opts = sqlx_opts.disable_statement_logging();
        } else {
            sqlx_opts = sqlx_opts.log_statements(options.sqlx_logging_level);
            if options.sqlx_slow_statements_logging_level != LevelFilter::Off {
                sqlx_opts = sqlx_opts.log_slow_statements(
                    options.sqlx_slow_statements_logging_level,
                    options.sqlx_slow_statements_logging_threshold,
                );
            }
        }
        // ...
        let lazy = options.connect_lazy;
        let mut pool_options = options.sqlx_pool_options();
        // ...
        let pool = if lazy {
            pool_options.connect_lazy_with(sqlx_opts)
        } else {
            pool_options
                .connect_with(sqlx_opts)
                .await
                .map_err(sqlx_error_to_conn_err)?
        };

impl Default for LogSettings {
    fn default() -> Self {
        LogSettings {
            statements_level: LevelFilter::Debug,
            slow_statements_level: LevelFilter::Warn,
            slow_statements_duration: Duration::from_secs(1),
        }
    }
}

 INFO sqlx::query: summary="select 1" db.statement="" rows_affected=1 rows_returned=1 elapsed=1.208667ms elapsed_secs=0.001208667

let db = Database::connect(build_connect_options(&lab_url("site-log"))).await?;
let sql = |s: &str| Statement::from_string(DbBackend::Postgres, s);

println!("--- a fast query ---");
db.execute(sql("select 1")).await?;
println!("--- a query that takes 1.2s ---");
db.execute(sql("select pg_sleep(1.2)")).await?;

println!("--- someone holds the only connection for 2.5s ---");
let txn = db.begin().await?;
tokio::spawn(async move {
  tokio::time::sleep(Duration::from_millis(2500)).await;
  txn.commit().await.unwrap();
});
let t = Instant::now();
db.execute(sql("select 2")).await?;
println!("waited {:?}", t.elapsed());

println!("--- held for longer than acquire_timeout (5s) ---");
let txn = db.begin().await?;
let t = Instant::now();
let err = db.execute(sql("select 3")).await.unwrap_err();
println!("after {:?}: {err}", t.elapsed());
println!("debug: {err:?}");
txn.rollback().await?;

--- a fast query ---
--- a query that takes 1.2s ---
 WARN sqlx::query: slow statement: execution time exceeded alert threshold summary="select pg_sleep(1.2)" db.statement="" rows_affected=1 rows_returned=1 elapsed=1.21419825s elapsed_secs=1.21419825 slow_threshold=1s
--- someone holds the only connection for 2.5s ---
 WARN sqlx::pool::acquire: acquired connection, but time to acquire exceeded slow threshold aquired_after_secs=2.504606334 slow_acquire_threshold_secs=2.0
waited 2.505586792s
--- held for longer than acquire_timeout (5s) ---
after 5.003468958s: Failed to acquire connection from pool: Connection pool timed out
debug: ConnectionAcquire(Timeout)

--- a fast query ---
DEBUG sqlx::query: summary="select 1" db.statement="" rows_affected=1 rows_returned=1 elapsed=559.542µs elapsed_secs=0.000559542
...
--- someone holds the only connection for 2.5s ---
DEBUG sqlx::query: summary="COMMIT" db.statement="" rows_affected=0 rows_returned=0 elapsed=1.136417ms elapsed_secs=0.001136417
 WARN sqlx::pool::acquire: acquired connection, but time to acquire exceeded slow threshold aquired_after_secs=2.505026833 slow_acquire_threshold_secs=2.0
DEBUG sqlx::query: summary="select 2" db.statement="" rows_affected=1 rows_returned=1 elapsed=1.453208ms elapsed_secs=0.001453208
...

pub(crate) fn sqlx_query(stmt: &Statement) -> sqlx::query::Query<'_, Postgres, SqlxValues> {
    let values = stmt
        .values
        .as_ref()
        .map_or(Values(Vec::new()), |values| values.clone());
    sqlx::query_with(&stmt.sql, SqlxValues(values))
}

for round in 1..=2 {
  println!("--- find_by_id(1).one(), round {round} ---");
  let m = Post::find_by_id(1_i64).one(&db).await?;
  // ...
}
for ids in [vec![1_i64, 2], vec![1, 2, 3]] {
  println!("--- filter(id.is_in({ids:?})).all() ---");
  let n = Post::find().filter(post::Column::Id.is_in(ids)).all(&db).await?.len();
  // ...
}
println!("--- execute_unprepared ---");
db.execute_unprepared("select 1").await?;
println!("--- begin / insert / commit ---");
let txn = db.begin().await?;
post::ActiveModel { id: Set(100), title: Set("in txn".into()) }.insert(&txn).await?;
txn.commit().await?;
// 然后在同一条连接上读 cached_statements_size() 和 pg_prepared_statements

--- find_by_id(1).one(), round 1 ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_1, "SELECT \"post\".\"id\", \"post\".\"title\" FROM \"post\" WHERE \"post\".\"id\" = $1 LIMIT $2"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 8/6), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
=> Some("post 1")
--- find_by_id(1).one(), round 2 ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 8/6), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
=> Some("post 1")
--- filter(id.is_in([1, 2])).all() ---
  ...
  [conn 1] C->S Parse(sqlx_s_2, "SELECT \"post\".\"id\", \"post\".\"title\" FROM \"post\" WHERE \"post\".\"id\" IN ($1, $2)"), Describe(S, sqlx_s_2), Sync
  ...
=> 2 rows
--- filter(id.is_in([1, 2, 3])).all() ---
  ...
  [conn 1] C->S Parse(sqlx_s_3, "SELECT \"post\".\"id\", \"post\".\"title\" FROM \"post\" WHERE \"post\".\"id\" IN ($1, $2, $3)"), Describe(S, sqlx_s_3), Sync
  ...
=> 3 rows
--- execute_unprepared ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("select 1")
  [conn 1] S->C RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
--- begin / insert / commit ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_4, "INSERT INTO \"post\" (\"id\", \"title\") VALUES ($1, $2) RETURNING \"id\", \"title\""), Describe(S, sqlx_s_4), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_4), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 8/6), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
cached_statements_size = 4, pg_prepared_statements = ["sqlx_s_1", "sqlx_s_2", "sqlx_s_3", "sqlx_s_4"]

    let db = db.into_schema_manager_connection();

    match db.get_database_backend() {
        DbBackend::Postgres => {
            let transaction = db.begin().await?;
            let manager = SchemaManager::new(&transaction);
            f(&manager).await?;
            transaction.commit().await
        }
        DbBackend::MySql | DbBackend::Sqlite => {
            let manager = SchemaManager::new(db);
            f(&manager).await
        }
    }

--- Migrator::up, first run ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_1, "CREATE TABLE IF NOT EXISTS \"seaql_migrations\" ( \"version\" varchar NOT NULL PRIMARY KEY, \"applied_at\" bigint NOT NULL )"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, NoData, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("CREATE TABLE"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, NoticeResponse, CommandComplete("CREATE TABLE"), CloseComplete, ReadyForQuery(T)
  ...
  [conn 1] C->S Parse(sqlx_s_2, "SELECT \"version\", \"applied_at\" FROM \"seaql_migrations\" ORDER BY \"version\" ASC"), Describe(S, sqlx_s_2), Sync
  ...
  [conn 1] C->S Parse(sqlx_s_3, "CREATE TABLE \"m_a\" ( \"id\" bigint PRIMARY KEY )"), Describe(S, sqlx_s_3), Sync
  ...
  [conn 1] C->S Parse(sqlx_s_4, "INSERT INTO \"seaql_migrations\" (\"version\", \"applied_at\") VALUES ($1, $2) RETURNING \"version\""), Describe(S, sqlx_s_4), Sync
  ...
  [conn 1] C->S Query("create table m_b (id int8 primary key)")
  [conn 1] S->C CommandComplete("CREATE TABLE"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_4), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 11), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
--- Migrator::up, second run ---
  ...
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 11/8), DataRow(列长 11/8), CommandComplete("SELECT 2"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  ...
seaql_migrations columns [("version", "character varying"), ("applied_at", "bigint")]
rows [("m1_create_a",), ("m2_create_b",)]

instance a: Err("Execution Error: error returned from database: duplicate key value violates unique constraint \"pg_type_typname_nsp_index\"")
instance b: Ok(())
rows [("m1_create_a",), ("m2_create_b",)]
instance a: Err("Execution Error: error returned from database: duplicate key value violates unique constraint \"pg_type_typname_nsp_index\"")
instance b: Ok(())
rows [("m1_create_a",), ("m2_create_b",)]
instance a: Ok(())
instance b: Err("Execution Error: error returned from database: duplicate key value violates unique constraint \"pg_type_typname_nsp_index\"")
rows [("m1_create_a",), ("m2_create_b",)]

    0ms size=2 idle=2 backends=[973, 974]
  511ms size=2 idle=2 backends=[973, 974]
 1014ms size=2 idle=2 backends=[973, 974]
 1519ms size=2 idle=2 backends=[973, 974]
 2024ms size=2 idle=2 backends=[976, 977]
 2527ms size=2 idle=2 backends=[976, 977]
 3031ms size=2 idle=2 backends=[976, 977]
 3537ms size=2 idle=2 backends=[976, 977]
 4045ms size=2 idle=1 backends=[978]

select count(*), current_setting('max_connections') from pg_stat_activity;

int8 into i32  : Query Error: error occurred while decoding column "id": mismatched types; Rust type `core::option::Option<i32>` (as SQL type `INT4`) is not compatible with SQL type `INT8`
NULL into String: Type Error: A null value was encountered while decoding "title"
unique violation: Execution Error: error returned from database: duplicate key value violates unique constraint "post_pkey"

before        : Some("post 1")
after ADD COLUMN        : Ok(Some("post 1"))
after ALTER COLUMN TYPE : Err("Query Error: error returned from database: cached plan must not change result type")
once more               : Err("Query Error: error returned from database: cached plan must not change result type")

p.delay_ms.store(100, Ordering::SeqCst);
let r = tokio::time::timeout(Duration::from_millis(150), db.begin()).await;
// ...
post::ActiveModel { id: Set(200), title: Set("after cancel".into()) }.insert(&db).await?;

--- timeout(150ms, db.begin()) ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
=> timed out, begin() future dropped
  [conn 1] C->S Sync
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] S->C ReadyForQuery(T)
--- an ordinary insert on the same pool ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_1, "INSERT INTO \"post\" (\"id\", \"title\") VALUES ($1, $2) RETURNING \"id\", \"title\""), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 8/12), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(T)
other session sees id 200: 0; pool backend: [(944, "idle in transaction")]
after the pool is closed, id 200 rows: 0

connect_timeout=Some(1s) acquire_timeout=Some(5s)
after 5.310339s: Connection Error: pool timed out while waiting for an open connection