时间戳

首页 > 解决方案 > 如何创建一个同时接受 sqlx 数据库池和事务的 actix-web 服务器？

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

问题描述

我正在尝试使用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);
    }
}

# 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"

编译输出

error[E0277]: the trait bound `Pool<Postgres>: Executor<'_>` is not satisfied
  --> src\main.rs:37:29
   |
17 | pub fn create_server<'a, E: 'static>(
   |        ------------- required by a bound in this
...
22 |     E: sqlx::Executor<'a, Database = sqlx::Postgres> + Copy,
   |        --------------------------------------------- required by this bound in `create_server`
...
37 |     create_server(listener, pool).unwrap().await
   |                             ^^^^ the trait `Executor<'_>` is not implemented for `Pool<Postgres>`
   |
   = help: the following implementations were found:
             <&Pool<DB> as Executor<'p>>

error[E0277]: the trait bound `Pool<Postgres>: Copy` is not satisfied
  --> src\main.rs:37:29
   |
17 | pub fn create_server<'a, E: 'static>(
   |        ------------- required by a bound in this
...
22 |     E: sqlx::Executor<'a, Database = sqlx::Postgres> + Copy,
   |                                                        ---- required by this bound in `create_server`
...
37 |     create_server(listener, pool).unwrap().await
   |                             ^^^^ the trait `Copy` is not implemented for `Pool<Postgres>`

标签： typesrustlifetimeactix-webrust-sqlx

解决方案

试图对 Executor 特征进行通用化有点矫枉过正。您可能应该只在测试中使用大小为 1 的池并手动调用Beginand 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");
}

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

// 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);
    }
}

推荐阅读