feat: impl using sqlx instead of diesel for fully async support

Cargo.toml

@@ -50,7 +50,8 @@ async-trait = "0.1.74"
 diesel = { version = "2.1.3", features = ["chrono", "mysql", "postgres", "sqlite", "r2d2"] }
 r2d2 = "0.8.10"
 
-sqlx = { version = "0.7.2", features = [ "runtime-tokio-rustls" , "postgres", "chrono", "migrate", "uuid", "json"], default-features = false }
+sqlx = { version = "0.7.2", features = [ "runtime-tokio" ,"tls-rustls", "postgres", "sqlite", "chrono", "migrate", "uuid", "json", "macros"], default-features = false }
+
 sea-orm = { version = "0.12.4", features = [ "sqlx-postgres", "runtime-tokio-rustls", "macros", "debug-print", "with-chrono" ], default-features = false }
 
 [build-dependencies]

src/apps/actix/server.rs

@@ -2,7 +2,7 @@ use actix_web::{web, App, HttpServer};
 
 pub async fn serve() -> std::io::Result<()> {
     // construct di
-    let user_component = crate::container::UserContainer::new();
+    let user_component = crate::container::UserContainer::new().await;
 
     let user_security_service = user_component.user_security_service.clone();
     let user_service = user_component.user_service.clone();

src/apps/axum/server.rs

@@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
 
 pub async fn serve() -> std::io::Result<()> {
     // construct di
-    let user_component = crate::container::UserContainer::new();
+    let user_component = crate::container::UserContainer::new().await;
 
     let app = Router::new()
         .route("/", get(health))

src/apps/cli/mod.rs

@@ -3,7 +3,7 @@ use chrono::Utc;
 use crate::users::User;
 
 pub async fn run() -> std::io::Result<()> {
-    let cmp = crate::container::UserContainer::new();
+    let cmp = crate::container::UserContainer::new().await;
     let user_service = cmp.user_service;
     let now = Utc::now().naive_utc();
     let user = User {

src/apps/warp/mod.rs

@@ -1,14 +1,13 @@
-use warp::Filter;
 use crate::core::QueryParamsImpl;
+use warp::Filter;
 
+pub mod error;
 pub mod health;
 pub mod users;
-pub mod error;
-
 
-pub fn routes() -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
+pub async fn routes() -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
     // construct di
-    let user_component = crate::container::UserContainer::new();
+    let user_component = crate::container::UserContainer::new().await;
     let user_service = user_component.user_service.clone();
 
     let index = warp::path::end()

src/bin/warp.rs

@@ -1,6 +1,6 @@
 #[tokio::main]
 async fn main() {
-    let routes = rwebapi::apps::warp::routes();
+    let routes = rwebapi::apps::warp::routes().await;
     let port = 8000;
     println!("Listenting {}", port);
     warp::serve(routes).run(([0, 0, 0, 0], port)).await;

src/container/mod.rs

@@ -8,7 +8,10 @@ pub struct UserContainer {
 }
 
 impl UserContainer {
-    pub fn new() -> Self {
+    pub async fn new() -> Self {
+        let sqlx_pool = crate::infra::db_conn().await;
+        let sqlx_user_repo: Arc<dyn UserRepo> =
+            Arc::new(crate::users::UserSqlxRepoImpl::new(sqlx_pool.clone()));
         let pool = Arc::new(crate::diesel_impl::db_pool());
         let user_repo: Arc<dyn UserRepo> = Arc::new(crate::diesel_impl::UserDieselImpl::new(pool));
 
@@ -20,7 +23,7 @@ impl UserContainer {
             });
 
         let user_service: Arc<dyn UserService> = Arc::new(UserServiceImpl {
-            user_repo: user_repo.clone(),
+            user_repo: sqlx_user_repo.clone(),
             user_security: user_security_service.clone(),
         });
         let user_auth_service: Arc<dyn UserAuthService> = Arc::new(UserAuthServiceImpl {
@@ -35,9 +38,3 @@ impl UserContainer {
         }
     }
 }
-
-impl Default for UserContainer {
-    fn default() -> Self {
-        Self::new()
-    }
-}

src/infra/db.rs

@@ -0,0 +1,36 @@
+#[cfg(feature = "postgres")]
+pub type DBConn = sqlx::postgres::PgPool;
+
+#[cfg(feature = "sqlite")]
+pub type DBConn = sqlx::sqlite::SqlitePool;
+
+#[cfg(feature = "mysql")]
+pub type DBConn = sqlx::mysql::MySqlPool;
+
+pub async fn db_conn() -> DBConn {
+    let database_url = std::env::var("DATABASE_URL").unwrap_or("/tmp/test_examples.db".to_string());
+    println!("Using Database {}", database_url);
+    #[cfg(feature = "sqlite")]
+    {
+        sqlx::sqlite::SqlitePoolOptions::new()
+            .connect(&database_url)
+            .await
+            .unwrap()
+    }
+
+    #[cfg(feature = "postgres")]
+    {
+        sqlx::postgres::PgPoolOptions::new()
+            .connect(&database_url)
+            .await
+            .unwrap()
+    }
+
+    #[cfg(feature = "mysql")]
+    {
+        sqlx::mysql::MySqlConnectOptions::new()
+            .connect(&database_url)
+            .await
+            .unwrap()
+    }
+}

src/infra/mod.rs

@@ -0,0 +1,2 @@
+mod db;
+pub use db::*;

src/lib.rs

@@ -5,4 +5,5 @@ pub mod container;
 pub mod core;
 pub mod diesel_impl;
 pub mod entity;
+pub mod infra;
 pub mod users;

src/users/mod.rs

@@ -1,5 +1,7 @@
 mod entity;
 mod services;
+mod user_repo;
 
 pub use entity::*;
 pub use services::*;
+pub use user_repo::*;

src/users/user_repo.rs

@@ -0,0 +1,72 @@
+use async_trait::async_trait;
+use tokio::join;
+
+use crate::core::{QueryParams, RepoResult, ResultPaging};
+
+use super::entity::{User, UserRepo, UserUpdate};
+
+pub struct UserSqlxRepoImpl {
+    pool: crate::infra::DBConn,
+}
+
+impl UserSqlxRepoImpl {
+    pub fn new(pool: crate::infra::DBConn) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl UserRepo for UserSqlxRepoImpl {
+    async fn get_all(&self, params: &dyn QueryParams) -> RepoResult<ResultPaging<User>> {
+        let pool = self.pool.clone();
+        let count_fut = sqlx::query!("SELECT COUNT(*) AS count FROM users").fetch_one(&pool);
+
+        let limit = params.limit();
+        let offset = params.offset();
+        let users_fut = sqlx::query_as!(
+            User,
+            r#"SELECT * FROM users ORDER BY id LIMIT ? OFFSET ?"#,
+            limit,
+            offset,
+        )
+        .fetch_all(&pool);
+        let (count, users) = join!(count_fut, users_fut);
+
+        return Ok(ResultPaging {
+            total: count.unwrap().count as i64,
+            items: users.unwrap(),
+        });
+    }
+
+    async fn find(&self, user_id: &str) -> RepoResult<User> {
+        let user = sqlx::query_as!(User, "SELECT * FROM users WHERE id = ?", user_id)
+            .fetch_one(&self.pool.clone())
+            .await
+            .unwrap();
+        Ok(user)
+    }
+
+    async fn find_by_email(&self, email: &str) -> RepoResult<User> {
+        let user = sqlx::query_as!(User, "SELECT * FROM users WHERE email = ?", email)
+            .fetch_one(&self.pool.clone())
+            .await
+            .unwrap();
+        return Ok(user);
+    }
+
+    async fn create(&self, user: &User) -> RepoResult<User> {
+        panic!("impl")
+    }
+
+    async fn update(&self, id: &str, update_user: &UserUpdate) -> RepoResult<User> {
+        panic!("impl")
+    }
+
+    async fn delete(&self, user_id: &str) -> RepoResult<()> {
+        sqlx::query!("DELETE FROM users WHERE id = ?", user_id)
+            .execute(&self.pool.clone())
+            .await
+            .unwrap();
+        Ok(())
+    }
+}