如何创建一个接受 sqlx 数据库池和事务的 actix-web 服务器？

Question

如何创建一个接受 sqlx 数据库池和事务的 actix-web 服务器？

Osk*_*son 2 types lifetime rust actix-web rust-sqlx

我正在尝试使用actix-web和sqlx设置一个 Web 应用程序，我可以在其中进行具有自己的 Web 服务器和数据库事务的测试。我尝试设置我的服务器创建，以便它接受数据库（Postgres）池或使用Executor特征的事务。虽然我在编译应用程序代码和测试时遇到了一些问题：

// main.rs

use std::net::TcpListener;

use actix_web::dev::Server;
use actix_web::{web, App, HttpServer, Responder};
use sqlx::PgPool;

async fn create_pool() -> PgPool {
    PgPool::connect("postgres://postgres:postgres@localhost:5432/postgres")
        .await
        .expect("Failed to create pool")
}

async fn index() -> impl Responder {
    "Hello World!"
}

pub fn create_server<'a, E: 'static>(
    listener: TcpListener,
    pool: E,
) -> Result<Server, std::io::Error>
where
    E: sqlx::Executor<'a, Database = sqlx::Postgres> + Copy,
{
    let server = HttpServer::new(move || App::new().data(pool).route("/", web::get().to(index)))
        .listen(listener)?
        .run();
    Ok(server)
}

pub async fn server(pool: PgPool) -> std::io::Result<()> {
    const PORT: usize = 8088;
    let listener =
        TcpListener::bind(format!("0.0.0.0:{}", PORT)).expect("Failed to create listener");

    println!("Running on port {}", PORT);

    create_server(listener, pool).unwrap().await
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let pool = create_pool().await;
    server(pool).await;
    Ok(())
}

#[cfg(test)]
pub mod tests {
    use super::*;
    use std::net::TcpListener;

    #[actix_rt::test]
    async fn test_foo() {
        let pool = create_pool().await;
        let mut transaction = pool.begin().await.expect("Failed to create transaction");

        let listener = TcpListener::bind("0.0.0.0:0").expect("Failed to create listener");
        let server = create_server(listener, &mut transaction).expect("Failed to create server");
        tokio::spawn(server);
    }
}

Run Code Online (Sandbox Code Playgroud)

# Cargo.toml

[package]
name = "sqlx-testing"
version = "0.1.0"
authors = ["Oskar"]
edition = "2018"

[dependencies]
actix-rt = "1.1.1"
actix-web = "3.3.2"
sqlx = { version = "0.4.2", default-features = false, features = ["postgres", "runtime-async-std-native-tls"] }
tokio = "0.2.22"

Run Code Online (Sandbox Code Playgroud)

编译输出

# Cargo.toml

[package]
name = "sqlx-testing"
version = "0.1.0"
authors = ["Oskar"]
edition = "2018"

[dependencies]
actix-rt = "1.1.1"
actix-web = "3.3.2"
sqlx = { version = "0.4.2", default-features = false, features = ["postgres", "runtime-async-std-native-tls"] }
tokio = "0.2.22"

Run Code Online (Sandbox Code Playgroud)

Answer 1

小智 6

试图对 Executor 特性进行通用化有点矫枉过正。您可能应该在测试中使用大小为 1 的池并手动调用Begin和ROLLBACK。

#[actix_rt::test]
async fn test_endpoint() {
    // build with only one connection
    let pool = PgPoolOptions::new()
        .max_connections(1)
        .connect("postgres://postgres:postgres@localhost:5432/postgres")
        .await
        .expect("pool failed");

    sqlx::query("BEGIN")
        .execute(&pool)
        .await
        .expect("BEGIN failed");
    let saved_pool = pool.clone();
    let listener = TcpListener::bind("0.0.0.0:0").expect("Failed to create listener");
    let server = HttpServer::new(move || 
    App::new().data(pool.clone()).service(one))
            .listen(listener)
            .expect("fail to bind")
            .run();
    tokio::spawn(server);

    // your test

    sqlx::query("ROLLBACK")
        .execute(&saved_pool)
        .await
        .expect("ROLLBACK failed");
}

Run Code Online (Sandbox Code Playgroud)

这样您就不必更改代码来处理您的测试

// main.rs
use actix_web::{get, web, App, HttpServer, Responder};
use sqlx::{postgres::PgPool, Row};
use std::net::TcpListener;

#[get("/one")]
async fn one(pool: web::Data<PgPool>) -> impl Responder {
    let row = sqlx::query("select 1 as id")
        .fetch_one(pool.get_ref())
        .await
        .unwrap();
    let one: i32 = row.try_get("id").unwrap();
    format!("{:?}", one)
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let pool = PgPool::connect("postgres://postgres:postgres@localhost:5432/postgres")
        .await
        .expect("Failed to create pool");
    const PORT: usize = 8088;
    let listener =
        TcpListener::bind(format!("0.0.0.0:{}", PORT)).expect("Failed to create listener");

    println!("Running on port {}", PORT);
    HttpServer::new(move || App::new().data(pool.clone()).service(one))
        .listen(listener)?
        .run()
        .await
}

#[cfg(test)]
pub mod tests {
    use super::*;
    use sqlx::postgres::PgPoolOptions;

    #[actix_rt::test]
    async fn test_endpoint() {
        // build with only one connection
        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect("postgres://postgres:postgres@localhost:5432/postgres")
            .await
            .expect("pool failed");

        sqlx::query("BEGIN")
            .execute(&pool)
            .await
            .expect("BEGIN failed");

        let saved_pool = pool.clone();

        let listener = TcpListener::bind("0.0.0.0:0").expect("Failed to create listener");
        let server = HttpServer::new(move || App::new().data(pool.clone()).service(one))
            .listen(listener)
            .expect("fail to bind")
            .run();
        tokio::spawn(server);

        // your test

        sqlx::query("ROLLBACK")
            .execute(&saved_pool)
            .await
            .expect("ROLLBACK failed");
    }

    #[actix_rt::test]
    async fn test_rollback() {
        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect("postgres://postgres:postgres@localhost:5432/postgres")
            .await
            .expect("pool failed");

        sqlx::query("BEGIN")
            .execute(&pool)
            .await
            .expect("BEGIN failed");

        sqlx::query("CREATE TABLE  IF NOT EXISTS test (id SERIAL, name TEXT)")
            .execute(&pool)
            .await
            .expect("CREATE TABLE test failed");

        sqlx::query("INSERT INTO test (name) VALUES ('bob')")
            .execute(&pool)
            .await
            .expect("INSERT test failed");

        let count: i64 = sqlx::query("SELECT COUNT(id) as count from test")
            .fetch_one(&pool)
            .await
            .expect("SELECT COUNT test failed")
            .try_get("count")
            .unwrap();
        sqlx::query("ROLLBACK")
            .execute(&pool)
            .await
            .expect("ROLLBACK failed");

        assert_eq!(count, 1);
    }

    #[actix_rt::test]
    async fn test_no_rollback() {
        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect("postgres://postgres:postgres@localhost:5432/postgres")
            .await
            .expect("pool failed");

        sqlx::query("CREATE TABLE  IF NOT EXISTS test2 (id SERIAL, name TEXT)")
            .execute(&pool)
            .await
            .expect("CREATE TABLE test failed");

        sqlx::query("INSERT INTO test2 (name) VALUES ('bob')")
            .execute(&pool)
            .await
            .expect("INSERT test failed");

        let count: i64 = sqlx::query("SELECT COUNT(id) as count from test2")
            .fetch_one(&pool)
            .await
            .expect("SELECT COUNT failed")
            .try_get("count")
            .unwrap();

        // this will failed the second time you run your test
        assert_eq!(count, 1);
    }
}

Run Code Online (Sandbox Code Playgroud)

Answer 2

kfe*_*v91 5

尝试创建一个可以接受两者的通用请求处理程序PgPool，但&mut Transaction事实证明太具有挑战性。幸运的是，您可以PgPool通过将实例限制为 1 个连接并BEGIN在将其传递给任何处理程序之前执行查询来使实例表现得像事务一样：

async fn get_transaction_pool() -> PgPool {
    let pool = PgPoolOptions::new()
         .max_connections(1)
         .connect("postgres://postgres:postgres@localhost:5432/postgres")
         .await
         .expect("Failed to create test pool.");

    sqlx::query("BEGIN")
         .execute(&pool)
         .await
         .expect("Failed to BEGIN transaction.");

    pool
}

Run Code Online (Sandbox Code Playgroud)

我发现将上面的内容抽象成它自己的TestTransaction结构很有用，如下所示：

struct TestTransaction {
    pool: web::Data<PgPool>,
}

impl TestTransaction {
    async fn begin() -> Self {
        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect("postgres://postgres:postgres@localhost:5432/postgres")
            .await
            .expect("Failed to connect to test pool.");

        sqlx::query("BEGIN")
            .execute(&pool)
            .await
            .expect("Failed to BEGIN transaction.");

        TestTransaction {
            pool: web::Data::new(pool),
        }
    }

    fn get_pool(&self) -> web::Data<PgPool> {
        self.pool.clone()
    }

    async fn rollback(&self) {
        sqlx::query("ROLLBACK")
            .execute(self.pool.as_ref())
            .await
            .expect("Failed to ROLLBACK transaction.");
    }
}

Run Code Online (Sandbox Code Playgroud)

此外，您不需要HttpServer在每个单元测试中启动，您可以直接按照这个简单的模板测试处理程序：

#[actix_rt::test]
async fn test_case() {
    let tx = TestTransaction::begin().await;

    let response = request_handler_func(tx.get_pool()).await;
    assert_eq!(response, "some expected value here");

    tx.rollback().await;
}

Run Code Online (Sandbox Code Playgroud)

以下是完整内容main.rs和一些评论：

use actix_web::{web, App, HttpServer};
use sqlx::{PgPool, Row};
use std::net::TcpListener;

async fn create_item(pool: web::Data<PgPool>) -> String {
    let id = sqlx::query("INSERT INTO items (label) VALUES ('label text') RETURNING id")
        .fetch_one(pool.as_ref())
        .await
        .expect("Failed to create item.")
        .get::<i64, _>("id");
    format!("created item with id {}", id)
}

async fn count_items(pool: web::Data<PgPool>) -> String {
    let count = sqlx::query("SELECT count(*) FROM items")
        .fetch_one(pool.as_ref())
        .await
        .expect("Failed to fetch item count.")
        .get::<i64, _>("count");
    format!("{} items in db", count)
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let pool = PgPool::connect("postgres://postgres:postgres@localhost:5432/postgres")
        .await
        .expect("Failed to create pool.");

    sqlx::query("CREATE TABLE IF NOT EXISTS items (id BIGSERIAL PRIMARY KEY, label TEXT)")
        .execute(&pool)
        .await
        .expect("Failed to create items table.");

    let listener = TcpListener::bind("0.0.0.0:8080").expect("Failed to create listener");

    println!("Listening on http://localhost:8080");
    println!("Try endpoints GET /create-item & GET /count-items");

    HttpServer::new(move || {
        App::new()
            .data(pool.clone())
            .route("/create-item", web::get().to(create_item))
            .route("/count-items", web::get().to(count_items))
    })
    .listen(listener)?
    .run()
    .await
}

#[cfg(test)]
pub mod tests {
    use super::*;
    use sqlx::postgres::PgPoolOptions;

    struct TestTransaction {
        pool: web::Data<PgPool>,
    }

    impl TestTransaction {
        async fn begin() -> Self {
            let pool = PgPoolOptions::new()
                .max_connections(1)
                .connect("postgres://postgres:postgres@localhost:5432/postgres")
                .await
                .expect("Failed to create test pool.");

            sqlx::query("BEGIN")
                .execute(&pool)
                .await
                .expect("Failed to BEGIN transaction.");

            // below 2 queries are necessary so that tests are always
            // run from within the same environment conditions, i.e.
            // the items table should be empty

            sqlx::query("DROP TABLE IF EXISTS items")
                .execute(&pool)
                .await
                .expect("Failed to drop test items table.");

            sqlx::query("CREATE TABLE IF NOT EXISTS items (id BIGSERIAL PRIMARY KEY, label TEXT)")
                .execute(&pool)
                .await
                .expect("Failed to create test items table.");

            TestTransaction {
                pool: web::Data::new(pool),
            }
        }

        fn get_pool(&self) -> web::Data<PgPool> {
            self.pool.clone()
        }

        async fn rollback(&self) {
            sqlx::query("ROLLBACK")
                .execute(self.pool.as_ref())
                .await
                .expect("Failed to ROLLBACK transaction.");
        }
    }

    // all tests below are run in parallel and are
    // isolated within their own transaction instances

    #[actix_rt::test]
    async fn create_and_count_1_items() {
        let tx = TestTransaction::begin().await;

        let response = create_item(tx.get_pool()).await;
        assert_eq!(response, "created item with id 1");

        let response = count_items(tx.get_pool()).await;
        assert_eq!(response, "1 items in db");

        tx.rollback().await;
    }

    #[actix_rt::test]
    async fn create_and_count_2_items() {
        let tx = TestTransaction::begin().await;

        let response = create_item(tx.get_pool()).await;
        assert_eq!(response, "created item with id 1");
        let response = create_item(tx.get_pool()).await;
        assert_eq!(response, "created item with id 2");

        let response = count_items(tx.get_pool()).await;
        assert_eq!(response, "2 items in db");

        tx.rollback().await;
    }

    #[actix_rt::test]
    async fn create_and_count_3_items() {
        let tx = TestTransaction::begin().await;

        let response = create_item(tx.get_pool()).await;
        assert_eq!(response, "created item with id 1");
        let response = create_item(tx.get_pool()).await;
        assert_eq!(response, "created item with id 2");
        let response = create_item(tx.get_pool()).await;
        assert_eq!(response, "created item with id 3");

        let response = count_items(tx.get_pool()).await;
        assert_eq!(response, "3 items in db");

        tx.rollback().await;
    }
}

Run Code Online (Sandbox Code Playgroud)

您当然可以使用以下命令运行测试cargo test，但您也可以cargo run在浏览器中运行并访问端点：

http://localhost:8080/create-item
http://localhost:8080/count-items

尽管这些端点修改了数据库，但如果您关闭服务器并尝试cargo test再次运行，测试仍然会通过！这是因为该TestTransaction结构有效地截断了函数items内的表begin，这使得所有单元测试都可重现，无论数据库中实际是什么，并且它在回滚的事务中安全地执行此操作，因此数据库本身中不会修改任何数据。

归档时间：	4 年，8 月前
查看次数：	1682 次
最近记录：	4 年，8 月前